features/lsp_discovery_dos_protection.feature

@@ -0,0 +1,103 @@
+@tdd_issue
+@tdd_issue_7161
+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
+
+  # ------------------------------------------------------------------
+  # Timeout enforcement (prevents indefinite traversal)
+  # ------------------------------------------------------------------
+
+  @tdd_issue_7161
+  Scenario: detect_directory_languages enforces timeout in outer walk loop
+    Given ldos a LanguageDiscovery instance
+    And ldos a mock directory and a clock that exceeds timeout immediately
+    When ldos I detect directory languages without explicit limits
+    Then ldos the result is a list
+
+  @tdd_issue_7161
+  Scenario: detect_directory_languages enforces timeout during file iteration
+    Given ldos a LanguageDiscovery instance
+    And ldos a mock directory and a clock that exceeds timeout after first file
+    When ldos I detect directory languages without explicit limits
+    Then ldos the result is a list

features/steps/lsp_discovery_dos_protection_steps.py

@@ -0,0 +1,249 @@
+"""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
+
+import time as _real_time
+from typing import Any
+from unittest.mock import patch
+
+from behave import given, then, when
+from behave.runner import Context
+
+from cleveragents.lsp.discovery import LanguageDiscovery
+
+
+class _LDOSFakeTime:
+    """Stand-in for the ``time`` module inside ``discovery.py``.
+
+    Returns successive pre-programmed values from ``time()`` so the
+    DoS-protection timeout branches in ``detect_directory_languages``
+    can be exercised deterministically. Other attributes pass through
+    to the real ``time`` module so unrelated code paths are unaffected.
+    """
+
+    def __init__(self, values: list[float]) -> None:
+        self._values = list(values)
+        self._idx = 0
+
+    def time(self) -> float:
+        if self._idx < len(self._values):
+            v = self._values[self._idx]
+            self._idx += 1
+            return v
+        return self._values[-1]
+
+    def __getattr__(self, name: str) -> Any:
+        return getattr(_real_time, name)
+
+
+# ── 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 and a clock that exceeds timeout immediately")
+def step_ldos_given_timeout_outer(context: Context) -> None:
+    """Mock ``os.walk`` and ``time`` so the outer-loop timeout fires on iter 1."""
+    walk_results = [
+        ("/fake/project", ["sub"], ["file1.py"]),
+        ("/fake/project/sub", [], ["file2.py"]),
+    ]
+    patcher_walk = patch(
+        "cleveragents.lsp.discovery.os.walk",
+        return_value=iter(walk_results),
+    )
+    patcher_walk.start()
+    context.add_cleanup(patcher_walk.stop)
+
+    # Sequence: start_time=0.0, first outer-loop elapsed=100.0 > 30.0 -> break.
+    fake = _LDOSFakeTime([0.0, 100.0])
+    patcher_time = patch("cleveragents.lsp.discovery.time", fake)
+    patcher_time.start()
+    context.add_cleanup(patcher_time.stop)
+
+
+@given("ldos a mock directory and a clock that exceeds timeout after first file")
+def step_ldos_given_timeout_inner(context: Context) -> None:
+    """Mock ``os.walk`` and ``time`` so the inner-file-loop timeout fires on file 2."""
+    walk_results = [
+        ("/fake/project", [], ["file1.py", "file2.py", "file3.py"]),
+    ]
+    patcher_walk = patch(
+        "cleveragents.lsp.discovery.os.walk",
+        return_value=iter(walk_results),
+    )
+    patcher_walk.start()
+    context.add_cleanup(patcher_walk.stop)
+
+    # Stub detect_file_language so file 1 processes cleanly and we reach file 2.
+    patcher_detect = patch.object(
+        LanguageDiscovery,
+        "detect_file_language",
+        return_value="python",
+    )
+    patcher_detect.start()
+    context.add_cleanup(patcher_detect.stop)
+
+    # Sequence: start_time=0.0; outer elapsed=1.0 (ok); file1 inner=1.0 (ok);
+    # file2 inner=100.0 > 30.0 -> warning + break.
+    fake = _LDOSFakeTime([0.0, 1.0, 1.0, 100.0])
+    patcher_time = patch("cleveragents.lsp.discovery.time", fake)
+    patcher_time.start()
+    context.add_cleanup(patcher_time.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)

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",