diff --git a/features/lsp_discovery_dos_protection.feature b/features/lsp_discovery_dos_protection.feature new file mode 100644 index 000000000..413b07276 --- /dev/null +++ b/features/lsp_discovery_dos_protection.feature @@ -0,0 +1,83 @@ +Feature: LSP language discovery DoS protection + Prevent resource exhaustion attacks via directory traversal limits. + Issue #7161: Resource exhaustion DoS in LSP language discovery. + + # ------------------------------------------------------------------ + # Depth limit protection (prevents deep nesting attacks) + # ------------------------------------------------------------------ + + @tdd_issue_7161 + Scenario: detect_directory_languages respects max_depth limit + Given ldos a LanguageDiscovery instance + And ldos a mock directory tree with depth 10 + When ldos I detect directory languages with max_depth 5 + Then ldos the result is a list + + @tdd_issue_7161 + Scenario: detect_directory_languages allows traversal within max_depth + Given ldos a LanguageDiscovery instance + And ldos a mock directory tree with depth 3 + When ldos I detect directory languages with max_depth 5 + Then ldos the result is a list + + # ------------------------------------------------------------------ + # File count limit protection (prevents wide directory attacks) + # ------------------------------------------------------------------ + + @tdd_issue_7161 + Scenario: detect_directory_languages respects max_files limit + Given ldos a LanguageDiscovery instance + And ldos a mock directory with 1000 files + When ldos I detect directory languages with max_files 100 + Then ldos the result is a list + + @tdd_issue_7161 + Scenario: detect_directory_languages allows processing within max_files + Given ldos a LanguageDiscovery instance + And ldos a mock directory with 50 files + When ldos I detect directory languages with max_files 100 + Then ldos the result is a list + + # ------------------------------------------------------------------ + # Symlink loop protection (prevents circular traversal) + # ------------------------------------------------------------------ + + @tdd_issue_7161 + Scenario: detect_directory_languages does not follow symlinks + Given ldos a LanguageDiscovery instance + And ldos a mock directory with symlinks + When ldos I detect directory languages + Then ldos the result is a list + + # ------------------------------------------------------------------ + # Parameter validation + # ------------------------------------------------------------------ + + @tdd_issue_7161 + Scenario: detect_directory_languages rejects invalid max_depth + Given ldos a LanguageDiscovery instance + When ldos I detect directory languages with max_depth 0 + Then ldos a ValueError is raised + + @tdd_issue_7161 + Scenario: detect_directory_languages rejects invalid max_files + Given ldos a LanguageDiscovery instance + When ldos I detect directory languages with max_files 0 + Then ldos a ValueError is raised + + @tdd_issue_7161 + Scenario: detect_directory_languages rejects invalid timeout + Given ldos a LanguageDiscovery instance + When ldos I detect directory languages with timeout 0.0 + Then ldos a ValueError is raised + + # ------------------------------------------------------------------ + # Default parameter values + # ------------------------------------------------------------------ + + @tdd_issue_7161 + Scenario: detect_directory_languages uses default limits + Given ldos a LanguageDiscovery instance + And ldos a mock directory with 100 files + When ldos I detect directory languages without explicit limits + Then ldos the result is a list diff --git a/features/steps/lsp_discovery_dos_protection_steps.py b/features/steps/lsp_discovery_dos_protection_steps.py new file mode 100644 index 000000000..6152241d4 --- /dev/null +++ b/features/steps/lsp_discovery_dos_protection_steps.py @@ -0,0 +1,171 @@ +"""Step definitions for lsp_discovery_dos_protection.feature. + +Tests for DoS protection in detect_directory_languages(): + - max_depth parameter prevents deep nesting attacks + - max_files parameter prevents wide directory attacks + - timeout parameter prevents indefinite traversal + - followlinks=False prevents symlink loop attacks + - Parameter validation rejects invalid values +""" + +from __future__ import annotations + +from unittest.mock import patch + +from behave import given, then, when +from behave.runner import Context + +from cleveragents.lsp.discovery import LanguageDiscovery + +# ── LanguageDiscovery instance ─────────────────────────────────────── + + +@given("ldos a LanguageDiscovery instance") +def step_ldos_given_discovery(context: Context) -> None: + context.ldos_discovery = LanguageDiscovery() + context.ldos_exception = None + context.ldos_result = None + + +# ── Mock directory trees ───────────────────────────────────────────── + + +@given("ldos a mock directory tree with depth {depth:d}") +def step_ldos_given_deep_tree(context: Context, depth: int) -> None: + """Create a mock os.walk that yields nested directories.""" + walk_results = [] + for d in range(depth): + root = "/fake" + "/level" * d + dirs = ["level"] if d < depth - 1 else [] + files = [f"file{d}.py"] + walk_results.append((root, dirs, files)) + + patcher_walk = patch( + "cleveragents.lsp.discovery.os.walk", + return_value=iter(walk_results), + ) + patcher_walk.start() + context.add_cleanup(patcher_walk.stop) + + # Mock open to avoid real file I/O + patcher_open = patch("builtins.open", side_effect=OSError("mocked")) + patcher_open.start() + context.add_cleanup(patcher_open.stop) + + +@given("ldos a mock directory with {count:d} files") +def step_ldos_given_wide_tree(context: Context, count: int) -> None: + """Create a mock os.walk that yields many files in one directory.""" + filenames = [f"file{i}.py" for i in range(count)] + walk_results = [("/fake/project", [], filenames)] + + patcher_walk = patch( + "cleveragents.lsp.discovery.os.walk", + return_value=iter(walk_results), + ) + patcher_walk.start() + context.add_cleanup(patcher_walk.stop) + + # Mock open to avoid real file I/O + patcher_open = patch("builtins.open", side_effect=OSError("mocked")) + patcher_open.start() + context.add_cleanup(patcher_open.stop) + + +@given("ldos a mock directory with symlinks") +def step_ldos_given_symlink_tree(context: Context) -> None: + """Create a mock os.walk that would loop with followlinks=True.""" + walk_results = [ + ("/fake/project", ["subdir"], ["file.py"]), + ("/fake/project/subdir", [], ["file2.py"]), + ] + + # Capture the followlinks argument + context.ldos_followlinks_arg = None + + def capture_walk(directory: str, followlinks: bool = True) -> None: + context.ldos_followlinks_arg = followlinks + return iter(walk_results) + + patcher_walk = patch( + "cleveragents.lsp.discovery.os.walk", + side_effect=capture_walk, + ) + patcher_walk.start() + context.add_cleanup(patcher_walk.stop) + + # Mock open to avoid real file I/O + patcher_open = patch("builtins.open", side_effect=OSError("mocked")) + patcher_open.start() + context.add_cleanup(patcher_open.stop) + + +# ── Detect directory languages with limits ─────────────────────────── + + +@when("ldos I detect directory languages with max_depth {max_depth:d}") +def step_ldos_when_detect_with_depth(context: Context, max_depth: int) -> None: + """Call detect_directory_languages with a specific max_depth.""" + try: + context.ldos_result = context.ldos_discovery.detect_directory_languages( + "/fake/project", + max_depth=max_depth, + ) + except ValueError as e: + context.ldos_exception = e + + +@when("ldos I detect directory languages with max_files {max_files:d}") +def step_ldos_when_detect_with_files(context: Context, max_files: int) -> None: + """Call detect_directory_languages with a specific max_files.""" + try: + context.ldos_result = context.ldos_discovery.detect_directory_languages( + "/fake/project", + max_files=max_files, + ) + except ValueError as e: + context.ldos_exception = e + + +@when("ldos I detect directory languages with timeout {timeout:f}") +def step_ldos_when_detect_with_timeout(context: Context, timeout: float) -> None: + """Call detect_directory_languages with a specific timeout.""" + try: + context.ldos_result = context.ldos_discovery.detect_directory_languages( + "/fake/project", + timeout=timeout, + ) + except ValueError as e: + context.ldos_exception = e + + +@when("ldos I detect directory languages without explicit limits") +def step_ldos_when_detect_default(context: Context) -> None: + """Call detect_directory_languages with default parameters.""" + context.ldos_result = context.ldos_discovery.detect_directory_languages( + "/fake/project", + ) + + +@when("ldos I detect directory languages") +def step_ldos_when_detect(context: Context) -> None: + """Call detect_directory_languages with default parameters.""" + context.ldos_result = context.ldos_discovery.detect_directory_languages( + "/fake/project", + ) + + +# ── Assertions ─────────────────────────────────────────────────────── + + +@then("ldos the result is a list") +def step_ldos_then_result_list(context: Context) -> None: + """Verify that the result is a list.""" + assert isinstance(context.ldos_result, list) + + +@then("ldos a ValueError is raised") +def step_ldos_then_value_error(context: Context) -> None: + """Verify that a ValueError was raised.""" + assert context.ldos_exception is not None + assert isinstance(context.ldos_exception, ValueError) diff --git a/src/cleveragents/lsp/discovery.py b/src/cleveragents/lsp/discovery.py index 5ba5b38c9..ef746666f 100644 --- a/src/cleveragents/lsp/discovery.py +++ b/src/cleveragents/lsp/discovery.py @@ -24,6 +24,7 @@ Based on ``docs/specification.md`` Resource Language Discovery from __future__ import annotations import os +import time from typing import Any import structlog @@ -204,26 +205,101 @@ class LanguageDiscovery: self._cache[file_path] = lang return lang - def detect_directory_languages(self, directory: str) -> list[str]: + def detect_directory_languages( + self, + directory: str, + max_depth: int = 50, + max_files: int = 10000, + timeout: float = 30.0, + ) -> list[str]: """Discover all languages present in a directory tree. Walks the directory, detects each file's language, and returns - a deduplicated sorted list. + a deduplicated sorted list. Enforces configurable limits on + traversal depth, file count, and execution time to prevent + resource exhaustion DoS attacks. Args: directory: Root directory to scan. + max_depth: Maximum directory depth to traverse (default: 50). + Prevents deep nesting attacks. + max_files: Maximum number of files to process (default: 10,000). + Prevents wide directory attacks. + timeout: Maximum execution time in seconds (default: 30.0). + Prevents indefinite traversal. Returns: Sorted list of unique language identifiers. """ + if max_depth < 1: + raise ValueError("max_depth must be >= 1") + if max_files < 1: + raise ValueError("max_files must be >= 1") + if timeout <= 0: + raise ValueError("timeout must be > 0") + languages: set[str] = set() + file_count = 0 + start_time = time.time() + try: - for root, _dirs, files in os.walk(directory): + for root, _dirs, files in os.walk(directory, followlinks=False): + # Check timeout + elapsed = time.time() - start_time + if elapsed > timeout: + logger.warning( + "lsp.discovery.timeout_exceeded", + directory=directory, + timeout=timeout, + elapsed=elapsed, + files_processed=file_count, + ) + break + + # Calculate current depth + depth = root[len(directory):].count(os.sep) + if depth > max_depth: + logger.warning( + "lsp.discovery.max_depth_exceeded", + directory=directory, + max_depth=max_depth, + current_depth=depth, + ) + break + for fname in files: + # Check file count limit + if file_count >= max_files: + logger.warning( + "lsp.discovery.max_files_exceeded", + directory=directory, + max_files=max_files, + files_processed=file_count, + ) + break + + # Check timeout again before processing each file + elapsed = time.time() - start_time + if elapsed > timeout: + logger.warning( + "lsp.discovery.timeout_exceeded", + directory=directory, + timeout=timeout, + elapsed=elapsed, + files_processed=file_count, + ) + break + fpath = os.path.join(root, fname) lang = self.detect_file_language(fpath) if lang != "plaintext": languages.add(lang) + file_count += 1 + + # Break outer loop if file count exceeded + if file_count >= max_files: + break + except OSError: logger.warning( "lsp.discovery.walk_error",