chore: import upstream snapshot with attribution

tests/security/test_path_validator.py

@@ -0,0 +1,504 @@
+"""
+Tests for PathValidator security module.
+"""
+
+import sys
+import tempfile
+from pathlib import Path
+
+import pytest
+
+from local_deep_research.security.path_validator import PathValidator
+
+
+class TestValidateSafePath:
+    """Tests for PathValidator.validate_safe_path()."""
+
+    @pytest.fixture
+    def temp_base_dir(self):
+        """Create a temporary base directory for tests."""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            yield Path(tmpdir)
+
+    def test_valid_relative_path(self, temp_base_dir):
+        """Accepts valid relative path."""
+        result = PathValidator.validate_safe_path(
+            "subdir/file.txt", temp_base_dir
+        )
+        assert result is not None
+        assert str(temp_base_dir) in str(result)
+
+    def test_valid_simple_filename(self, temp_base_dir):
+        """Accepts simple filename."""
+        result = PathValidator.validate_safe_path("file.txt", temp_base_dir)
+        assert result is not None
+        assert result.name == "file.txt"
+
+    def test_path_traversal_blocked(self, temp_base_dir):
+        """Blocks path traversal attempts."""
+        with pytest.raises(ValueError) as exc_info:
+            PathValidator.validate_safe_path("../etc/passwd", temp_base_dir)
+        assert "traversal" in str(exc_info.value).lower()
+
+    def test_path_traversal_double_dots(self, temp_base_dir):
+        """Blocks multiple .. components."""
+        with pytest.raises(ValueError) as exc_info:
+            PathValidator.validate_safe_path("../../secret.txt", temp_base_dir)
+        assert "traversal" in str(exc_info.value).lower()
+
+    def test_path_traversal_mixed(self, temp_base_dir):
+        """Blocks mixed path with traversal."""
+        with pytest.raises(ValueError) as exc_info:
+            PathValidator.validate_safe_path(
+                "valid/../../../escape.txt", temp_base_dir
+            )
+        assert "traversal" in str(exc_info.value).lower()
+
+    def test_empty_path_rejected(self, temp_base_dir):
+        """Rejects empty path."""
+        with pytest.raises(ValueError) as exc_info:
+            PathValidator.validate_safe_path("", temp_base_dir)
+        assert "invalid" in str(exc_info.value).lower()
+
+    def test_none_path_rejected(self, temp_base_dir):
+        """Rejects None path."""
+        with pytest.raises(ValueError):
+            PathValidator.validate_safe_path(None, temp_base_dir)
+
+    def test_required_extensions(self, temp_base_dir):
+        """Validates required file extensions."""
+        result = PathValidator.validate_safe_path(
+            "config.json",
+            temp_base_dir,
+            required_extensions=(".json", ".yaml"),
+        )
+        assert result.suffix == ".json"
+
+    def test_wrong_extension_rejected(self, temp_base_dir):
+        """Rejects file with wrong extension."""
+        with pytest.raises(ValueError) as exc_info:
+            PathValidator.validate_safe_path(
+                "script.py",
+                temp_base_dir,
+                required_extensions=(".json", ".yaml"),
+            )
+        assert "allowed" in str(exc_info.value).lower()
+
+    def test_whitespace_stripped(self, temp_base_dir):
+        """Strips whitespace from path."""
+        result = PathValidator.validate_safe_path("  file.txt  ", temp_base_dir)
+        assert result.name == "file.txt"
+
+
+class TestValidateLocalFilesystemPath:
+    """Tests for PathValidator.validate_local_filesystem_path()."""
+
+    @pytest.fixture
+    def temp_safe_dir(self):
+        """Create a temporary safe directory."""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            yield Path(tmpdir)
+
+    def test_valid_absolute_path(self, temp_safe_dir):
+        """Accepts valid absolute path."""
+        test_path = str(temp_safe_dir / "subdir")
+        result = PathValidator.validate_local_filesystem_path(test_path)
+        assert result is not None
+
+    @pytest.mark.skipif(
+        Path.home() == Path("/root"),
+        reason="Skipping in Docker/CI where home is /root (restricted directory)",
+    )
+    def test_home_expansion(self):
+        """Expands ~ to home directory."""
+        result = PathValidator.validate_local_filesystem_path("~/Documents")
+        assert str(result).startswith(str(Path.home()))
+
+    def test_null_byte_rejected(self):
+        """Rejects path with null bytes."""
+        with pytest.raises(ValueError) as exc_info:
+            PathValidator.validate_local_filesystem_path("/path/with\x00null")
+        assert "null" in str(exc_info.value).lower()
+
+    def test_control_characters_rejected(self):
+        """Rejects path with control characters."""
+        with pytest.raises(ValueError) as exc_info:
+            PathValidator.validate_local_filesystem_path(
+                "/path/with\x01control"
+            )
+        assert "control" in str(exc_info.value).lower()
+
+    def test_empty_path_rejected(self):
+        """Rejects empty path."""
+        with pytest.raises(ValueError):
+            PathValidator.validate_local_filesystem_path("")
+
+    def test_none_path_rejected(self):
+        """Rejects None path."""
+        with pytest.raises(ValueError):
+            PathValidator.validate_local_filesystem_path(None)
+
+    @pytest.mark.skipif(sys.platform == "win32", reason="Unix-specific test")
+    def test_etc_blocked(self):
+        """Blocks access to /etc directory."""
+        with pytest.raises(ValueError) as exc_info:
+            PathValidator.validate_local_filesystem_path("/etc/passwd")
+        assert "system" in str(exc_info.value).lower()
+
+    @pytest.mark.skipif(sys.platform == "win32", reason="Unix-specific test")
+    def test_proc_blocked(self):
+        """Blocks access to /proc directory."""
+        with pytest.raises(ValueError) as exc_info:
+            PathValidator.validate_local_filesystem_path("/proc/1/status")
+        assert "system" in str(exc_info.value).lower()
+
+    @pytest.mark.skipif(sys.platform == "win32", reason="Unix-specific test")
+    def test_sys_blocked(self):
+        """Blocks access to /sys directory."""
+        with pytest.raises(ValueError) as exc_info:
+            PathValidator.validate_local_filesystem_path("/sys/class")
+        assert "system" in str(exc_info.value).lower()
+
+    @pytest.mark.skipif(sys.platform == "win32", reason="Unix-specific test")
+    def test_dev_blocked(self):
+        """Blocks access to /dev directory."""
+        with pytest.raises(ValueError) as exc_info:
+            PathValidator.validate_local_filesystem_path("/dev/null")
+        assert "system" in str(exc_info.value).lower()
+
+    @pytest.mark.skipif(sys.platform == "win32", reason="Unix-specific test")
+    def test_root_blocked(self):
+        """Blocks access to /root directory.
+
+        Also a regression guard for #3090: this must surface as a clean
+        ValueError. The removed pre-resolve is_symlink() lstat() raised an
+        uncaught PermissionError for /root/.bashrc on non-root hosts (where it
+        escaped to a Flask 500); pytest.raises(ValueError) here would not catch
+        that, so this test fails loudly if the EACCES leak ever returns.
+        """
+        with pytest.raises(ValueError) as exc_info:
+            PathValidator.validate_local_filesystem_path("/root/.bashrc")
+        assert "system" in str(exc_info.value).lower()
+
+    @pytest.mark.skipif(
+        sys.platform == "win32",
+        reason="symlink creation needs privileges on Windows",
+    )
+    def test_symlinked_directory_is_permitted(self, temp_safe_dir):
+        """Regression for #3090: a symlinked directory is accepted.
+
+        The removed pre-resolve symlink check rejected ALL symlinks, breaking
+        legitimate setups (Docker/k8s bind mounts, macOS /tmp -> /private/tmp).
+        Containment is still enforced by resolve()+restricted_dirs.
+        """
+        real = temp_safe_dir / "real"
+        real.mkdir()
+        link = temp_safe_dir / "link"
+        link.symlink_to(real, target_is_directory=True)
+
+        result = PathValidator.validate_local_filesystem_path(
+            str(link), restricted_dirs=[]
+        )
+        assert result is not None
+
+    @pytest.mark.skipif(sys.platform == "win32", reason="Unix-specific test")
+    def test_symlink_escaping_to_system_dir_still_blocked(self, temp_safe_dir):
+        """A symlink that resolves into a restricted system dir is still blocked:
+        resolve() follows it and the restricted-dir check catches the target.
+        This is the real symlink control (vs. the removed leaf-only lstat).
+        """
+        link = temp_safe_dir / "escape"
+        link.symlink_to("/etc", target_is_directory=True)
+
+        with pytest.raises(ValueError) as exc_info:
+            PathValidator.validate_local_filesystem_path(str(link / "passwd"))
+        assert "system" in str(exc_info.value).lower()
+
+    def test_custom_restricted_dirs(self, temp_safe_dir):
+        """Respects custom restricted directories."""
+        restricted = [temp_safe_dir]
+        with pytest.raises(ValueError):
+            PathValidator.validate_local_filesystem_path(
+                str(temp_safe_dir / "subdir"),
+                restricted_dirs=restricted,
+            )
+
+    def test_empty_restricted_dirs_allows_all(self, temp_safe_dir):
+        """Empty restricted dirs allows all paths."""
+        result = PathValidator.validate_local_filesystem_path(
+            str(temp_safe_dir),
+            restricted_dirs=[],
+        )
+        assert result is not None
+
+    def test_restricted_dir_block_does_not_log_user_path(self, temp_safe_dir):
+        """The restricted-dir block must log WHICH restricted dir was hit, not
+        the user's submitted path — a resolved local path can contain a
+        username."""
+        from loguru import logger
+
+        secret_segment = "alice-private-7f3a2b"
+        target = str(temp_safe_dir / secret_segment / "file.txt")
+
+        # The package disables loguru by default (__init__.py); enable it so the
+        # error emitted from inside the module reaches our sink.
+        logger.enable("local_deep_research")
+        logged: list[str] = []
+        sink_id = logger.add(
+            lambda m: logged.append(m.record["message"]), level="ERROR"
+        )
+        try:
+            with pytest.raises(ValueError):
+                PathValidator.validate_local_filesystem_path(
+                    target, restricted_dirs=[temp_safe_dir]
+                )
+        finally:
+            logger.remove(sink_id)
+            logger.disable("local_deep_research")
+
+        blocked = [m for m in logged if "restricted directory" in m.lower()]
+        assert blocked, "expected a restricted-dir block error log"
+        assert all(secret_segment not in m for m in blocked)
+
+
+class TestValidateConfigPath:
+    """Tests for PathValidator.validate_config_path()."""
+
+    @pytest.fixture
+    def temp_config_dir(self):
+        """Create temp config directory with test files."""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            config_file = Path(tmpdir) / "settings.json"
+            config_file.write_text('{"key": "value"}')
+            yaml_file = Path(tmpdir) / "config.yaml"
+            yaml_file.write_text("key: value")
+            yield Path(tmpdir)
+
+    def test_valid_relative_config_path(self, temp_config_dir):
+        """Accepts valid relative config path."""
+        result = PathValidator.validate_config_path(
+            "settings.json",
+            config_root=str(temp_config_dir),
+        )
+        assert result.exists()
+
+    @pytest.mark.skip(
+        reason="Absolute paths to temp dirs not supported by current implementation"
+    )
+    def test_valid_absolute_config_path(self, temp_config_dir):
+        """Accepts valid absolute config path."""
+        # NOTE: Current implementation of validate_config_path uses safe_join("/", path)
+        # for absolute paths, which doesn't work with arbitrary temp directories.
+        # This test would need the implementation to be updated to handle
+        # arbitrary absolute paths or config_root parameter for absolute paths.
+        # audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: REVISIT).
+        config_file = temp_config_dir / "settings.json"
+        result = PathValidator.validate_config_path(
+            str(config_file),
+            config_root=str(temp_config_dir),
+        )
+        assert result.exists()
+
+    def test_traversal_blocked(self, temp_config_dir):
+        """Blocks path traversal in config path."""
+        with pytest.raises(ValueError) as exc_info:
+            PathValidator.validate_config_path(
+                "../../../etc/passwd",
+                config_root=str(temp_config_dir),
+            )
+        assert "traversal" in str(exc_info.value).lower()
+
+    def test_null_bytes_rejected(self, temp_config_dir):
+        """Rejects null bytes in config path."""
+        with pytest.raises(ValueError, match="Null bytes"):
+            PathValidator.validate_config_path(
+                "set\x00tings.json",
+                config_root=str(temp_config_dir),
+            )
+
+    def test_etc_prefix_blocked(self, temp_config_dir):
+        """Blocks paths starting with /etc/."""
+        with pytest.raises(ValueError) as exc_info:
+            PathValidator.validate_config_path("/etc/passwd")
+        assert "restricted" in str(exc_info.value).lower()
+
+    def test_proc_prefix_blocked(self, temp_config_dir):
+        """Blocks paths starting with /proc/."""
+        with pytest.raises(ValueError) as exc_info:
+            PathValidator.validate_config_path("/proc/1/status")
+        assert "restricted" in str(exc_info.value).lower()
+
+    def test_wrong_extension_blocked(self, temp_config_dir):
+        """Blocks config files with wrong extension."""
+        with pytest.raises(ValueError) as exc_info:
+            PathValidator.validate_config_path(
+                "script.py",
+                config_root=str(temp_config_dir),
+            )
+        assert (
+            "allowed" in str(exc_info.value).lower()
+            or "type" in str(exc_info.value).lower()
+        )
+
+    def test_valid_yaml_extension(self, temp_config_dir):
+        """Accepts YAML config files."""
+        result = PathValidator.validate_config_path(
+            "config.yaml",
+            config_root=str(temp_config_dir),
+        )
+        assert result.suffix == ".yaml"
+
+
+class TestValidateModelPath:
+    """Tests for PathValidator.validate_model_path()."""
+
+    @pytest.fixture
+    def temp_model_dir(self):
+        """Create temp model directory with test files."""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            model_file = Path(tmpdir) / "model.gguf"
+            model_file.write_text("fake model content")
+            yield Path(tmpdir)
+
+    def test_valid_model_path(self, temp_model_dir):
+        """Accepts valid model path."""
+        result = PathValidator.validate_model_path(
+            "model.gguf",
+            model_root=str(temp_model_dir),
+        )
+        assert result.exists()
+
+    def test_model_not_found(self, temp_model_dir):
+        """Raises error for non-existent model."""
+        with pytest.raises(ValueError) as exc_info:
+            PathValidator.validate_model_path(
+                "missing.gguf",
+                model_root=str(temp_model_dir),
+            )
+        assert "not found" in str(exc_info.value).lower()
+
+    def test_model_path_is_directory(self, temp_model_dir):
+        """Raises error when path is directory."""
+        subdir = temp_model_dir / "subdir"
+        subdir.mkdir()
+        with pytest.raises(ValueError) as exc_info:
+            PathValidator.validate_model_path(
+                "subdir",
+                model_root=str(temp_model_dir),
+            )
+        assert "not a file" in str(exc_info.value).lower()
+
+    def test_model_traversal_blocked(self, temp_model_dir):
+        """Blocks path traversal in model path."""
+        with pytest.raises(ValueError):
+            PathValidator.validate_model_path(
+                "../../../etc/passwd",
+                model_root=str(temp_model_dir),
+            )
+
+
+class TestValidateDataPath:
+    """Tests for PathValidator.validate_data_path()."""
+
+    @pytest.fixture
+    def temp_data_dir(self):
+        """Create temp data directory."""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            yield Path(tmpdir)
+
+    def test_valid_data_path(self, temp_data_dir):
+        """Accepts valid data path."""
+        result = PathValidator.validate_data_path(
+            "data/file.txt",
+            str(temp_data_dir),
+        )
+        assert result is not None
+
+    def test_data_path_traversal_blocked(self, temp_data_dir):
+        """Blocks path traversal in data path."""
+        with pytest.raises(ValueError):
+            PathValidator.validate_data_path(
+                "../secret.txt",
+                str(temp_data_dir),
+            )
+
+
+class TestPathValidatorConstants:
+    """Tests for PathValidator constants."""
+
+    def test_safe_path_pattern_exists(self):
+        """SAFE_PATH_PATTERN is defined."""
+        assert PathValidator.SAFE_PATH_PATTERN is not None
+
+    def test_config_extensions_defined(self):
+        """CONFIG_EXTENSIONS is defined."""
+        assert PathValidator.CONFIG_EXTENSIONS is not None
+        assert ".json" in PathValidator.CONFIG_EXTENSIONS
+        assert ".yaml" in PathValidator.CONFIG_EXTENSIONS
+        assert ".yml" in PathValidator.CONFIG_EXTENSIONS
+
+
+class TestSecurityScenarios:
+    """Integration tests for security scenarios."""
+
+    @pytest.fixture
+    def temp_base_dir(self):
+        """Create a temporary base directory."""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            yield Path(tmpdir)
+
+    def test_unicode_normalization_attack(self, temp_base_dir):
+        """Unicode look-alike traversal is rejected.
+
+        Full-width periods (U+FF0E '．') NFKC-normalize to '.', which a
+        downstream consumer might apply silently. PathValidator rejects
+        the input before that happens.
+        """
+        # audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: REVISIT).
+        with pytest.raises(ValueError, match="unicode traversal"):
+            PathValidator.validate_safe_path(
+                "．．/etc/passwd",  # Full-width periods
+                temp_base_dir,
+            )
+
+    def test_url_encoded_traversal(self, temp_base_dir):
+        """URL-encoded path traversal is rejected.
+
+        Some downstream consumers urldecode strings before joining them
+        into paths; werkzeug.safe_join's literal-'..' check can't see
+        through '%2e%2e'. PathValidator catches this earlier.
+        """
+        # audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: REVISIT).
+        with pytest.raises(ValueError, match="encoded traversal"):
+            PathValidator.validate_safe_path(
+                "%2e%2e/etc/passwd",
+                temp_base_dir,
+            )
+
+    def test_double_encoded_traversal(self, temp_base_dir):
+        """Double-encoded path traversal is rejected.
+
+        '%252e%252e' decodes once to '%2e%2e' and again to '..';
+        detection runs two levels of unquote() to catch this layered
+        encoding.
+        """
+        # audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: REVISIT).
+        with pytest.raises(ValueError, match="encoded traversal"):
+            PathValidator.validate_safe_path(
+                "%252e%252e/etc/passwd",
+                temp_base_dir,
+            )
+
+    def test_symlink_escape_prevention(self, temp_base_dir):
+        """Path resolution prevents symlink escapes."""
+        # Create a symlink pointing outside base dir
+        symlink_path = temp_base_dir / "escape"
+        try:
+            symlink_path.symlink_to("/etc")
+            # Attempting to access via symlink should work for the symlink itself
+            # but not for paths through it
+            result = PathValidator.validate_safe_path("escape", temp_base_dir)
+            # The result should be within base_dir (the symlink itself)
+            assert str(temp_base_dir) in str(result)
+        except OSError:
+            pytest.skip("Symlink creation not supported")