From d08084236a13a850cc311a507521263c467108b9 Mon Sep 17 00:00:00 2001 From: CleverThis Date: Sun, 19 Apr 2026 09:18:44 +0000 Subject: [PATCH 1/2] fix(lsp/runtime): add workspace path containment check to _read_file LspRuntime._read_file had no check that the requested file path was contained within the workspace directory, allowing path traversal attacks to read arbitrary files on the filesystem. Changes: - Add _workspace_paths dict to LspRuntime to track per-server workspace roots - start_server now stores the resolved workspace path for each server - stop_server and stop_all clean up workspace path entries - _read_file now accepts an optional workspace_path parameter; when provided, it resolves both paths and raises LspError if the file is outside the workspace (prevents ../../etc/passwd style attacks) - get_diagnostics, get_completions, get_hover, get_definitions all pass the registered workspace path to _read_file - Add features/lsp_path_containment.feature with 10 BDD scenarios - Add features/steps/lsp_path_containment_steps.py step definitions ISSUES CLOSED: #10490 --- features/lsp_path_containment.feature | 83 +++++ features/steps/lsp_path_containment_steps.py | 317 +++++++++++++++++++ src/cleveragents/lsp/runtime.py | 64 +++- 3 files changed, 456 insertions(+), 8 deletions(-) create mode 100644 features/lsp_path_containment.feature create mode 100644 features/steps/lsp_path_containment_steps.py diff --git a/features/lsp_path_containment.feature b/features/lsp_path_containment.feature new file mode 100644 index 000000000..6b2261fad --- /dev/null +++ b/features/lsp_path_containment.feature @@ -0,0 +1,83 @@ +Feature: LspRuntime workspace path containment + As a security-conscious platform + I need LspRuntime._read_file to enforce workspace path containment + So that path traversal attacks cannot read files outside the workspace + + # ── _read_file static method containment ────────────────────────── + + Scenario: read_file allows a file inside the workspace + Given lspc I have a temp workspace directory + And lspc I have a file inside the workspace with content "safe content" + When lspc I call read_file with the workspace path + Then lspc the file content should be "safe content" + And lspc no error should be raised + + Scenario: read_file blocks a file outside the workspace + Given lspc I have a temp workspace directory + And lspc I have a file outside the workspace + When lspc I call read_file with the workspace path + Then lspc an LspError should be raised with message containing "outside workspace" + + Scenario: read_file blocks path traversal using dot-dot segments + Given lspc I have a temp workspace directory + And lspc I have a file outside the workspace + When lspc I call read_file with a traversal path and the workspace path + Then lspc an LspError should be raised with message containing "outside workspace" + + Scenario: read_file without workspace path has no containment check + Given lspc I have a file outside the workspace + When lspc I call read_file without a workspace path + Then lspc no error should be raised + + # ── get_diagnostics containment ──────────────────────────────────── + + Scenario: get_diagnostics blocks file outside workspace + Given lspc I have a temp workspace directory + And lspc I have a file outside the workspace + And lspc I create an LspRuntime with a healthy mock server "local/pyright" and workspace + When lspc I try to get diagnostics for "local/pyright" on the outside file + Then lspc an LspError should be raised with message containing "outside workspace" + + Scenario: get_diagnostics allows file inside workspace + Given lspc I have a temp workspace directory + And lspc I have a file inside the workspace with content "x = 1" + And lspc I create an LspRuntime with a healthy mock server "local/pyright" and workspace + When lspc I get diagnostics for "local/pyright" on the inside file + Then lspc diagnostics should be returned as a list + And lspc no error should be raised + + # ── get_completions containment ──────────────────────────────────── + + Scenario: get_completions blocks file outside workspace + Given lspc I have a temp workspace directory + And lspc I have a file outside the workspace + And lspc I create an LspRuntime with a healthy mock server "local/pyright" and workspace + When lspc I try to get completions for "local/pyright" on the outside file at line 1 column 1 + Then lspc an LspError should be raised with message containing "outside workspace" + + # ── get_hover containment ────────────────────────────────────────── + + Scenario: get_hover blocks file outside workspace + Given lspc I have a temp workspace directory + And lspc I have a file outside the workspace + And lspc I create an LspRuntime with a healthy mock server "local/pyright" and workspace + When lspc I try to get hover for "local/pyright" on the outside file at line 1 column 1 + Then lspc an LspError should be raised with message containing "outside workspace" + + # ── get_definitions containment ──────────────────────────────────── + + Scenario: get_definitions blocks file outside workspace + Given lspc I have a temp workspace directory + And lspc I have a file outside the workspace + And lspc I create an LspRuntime with a healthy mock server "local/pyright" and workspace + When lspc I try to get definitions for "local/pyright" on the outside file at line 1 column 1 + Then lspc an LspError should be raised with message containing "outside workspace" + + # ── workspace path not registered ───────────────────────────────── + + Scenario: get_diagnostics without registered workspace has no containment check + Given lspc I have a file outside the workspace + And lspc I create an LspRuntime with a healthy mock server "local/pyright" without workspace + When lspc I get diagnostics for "local/pyright" on the outside file + Then lspc diagnostics should be returned as a list + And lspc no error should be raised diff --git a/features/steps/lsp_path_containment_steps.py b/features/steps/lsp_path_containment_steps.py new file mode 100644 index 000000000..7f6e074e0 --- /dev/null +++ b/features/steps/lsp_path_containment_steps.py @@ -0,0 +1,317 @@ +"""Step definitions for lsp_path_containment.feature. + +Tests workspace path containment in LspRuntime._read_file to prevent +path traversal attacks. Uses the ``lspc`` step prefix to avoid +Behave AmbiguousStep errors. +""" + +from __future__ import annotations + +import os +import tempfile +from unittest.mock import MagicMock + +from behave import given, then, when +from behave.runner import Context + +from cleveragents.lsp.errors import LspError +from cleveragents.lsp.lifecycle import LspLifecycleManager +from cleveragents.lsp.runtime import LspRuntime + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_mock_client() -> MagicMock: + """Create a mock LSP client with the methods runtime calls.""" + client = MagicMock(name="mock_lsp_client") + client.did_open = MagicMock() + client.did_close = MagicMock() + client.get_diagnostics = MagicMock(return_value=[]) + client.get_completions = MagicMock(return_value=[]) + client.get_hover = MagicMock(return_value=None) + client.get_definitions = MagicMock(return_value=[]) + return client + + +# --------------------------------------------------------------------------- +# Given steps +# --------------------------------------------------------------------------- + + +@given("lspc I have a temp workspace directory") +def step_lspc_create_workspace(context: Context) -> None: + workspace = tempfile.mkdtemp(prefix="lspc_workspace_") + context.lspc_workspace = workspace + context.lspc_error = None + + def cleanup() -> None: + import shutil + if os.path.exists(workspace): + shutil.rmtree(workspace, ignore_errors=True) + + context.add_cleanup(cleanup) + + +@given('lspc I have a file inside the workspace with content "{content}"') +def step_lspc_create_inside_file(context: Context, content: str) -> None: + fd, path = tempfile.mkstemp( + suffix=".py", + dir=context.lspc_workspace, + prefix="inside_", + ) + with os.fdopen(fd, "w") as f: + f.write(content) + context.lspc_inside_file = path + + def cleanup() -> None: + if os.path.exists(path): + os.unlink(path) + + context.add_cleanup(cleanup) + + +@given("lspc I have a file outside the workspace") +def step_lspc_create_outside_file(context: Context) -> None: + fd, path = tempfile.mkstemp(suffix=".py", prefix="outside_") + with os.fdopen(fd, "w") as f: + f.write("outside content") + context.lspc_outside_file = path + context.lspc_error = None + + def cleanup() -> None: + if os.path.exists(path): + os.unlink(path) + + context.add_cleanup(cleanup) + + +@given( + 'lspc I create an LspRuntime with a healthy mock server "{name}" and workspace' +) +def step_lspc_create_runtime_with_workspace(context: Context, name: str) -> None: + mock_client = _make_mock_client() + + mock_lifecycle = MagicMock(spec=LspLifecycleManager) + mock_lifecycle.health_check = MagicMock(return_value=True) + mock_lifecycle.get_client = MagicMock(return_value=mock_client) + mock_lifecycle.start_server = MagicMock() + + runtime = LspRuntime(lifecycle_manager=mock_lifecycle) + # Register the workspace path by calling start_server + # We need to mock the registry lookup too + from cleveragents.lsp.models import LspServerConfig + from cleveragents.lsp.registry import LspRegistry + + registry = LspRegistry() + config = LspServerConfig(name=name, command="echo", languages=["python"]) + registry.register(config) + + runtime = LspRuntime(registry=registry, lifecycle_manager=mock_lifecycle) + runtime.start_server(name, context.lspc_workspace) + + context.lspc_runtime = runtime + context.lspc_mock_client = mock_client + context.lspc_error = None + + +@given( + 'lspc I create an LspRuntime with a healthy mock server "{name}" without workspace' +) +def step_lspc_create_runtime_without_workspace(context: Context, name: str) -> None: + mock_client = _make_mock_client() + + mock_lifecycle = MagicMock(spec=LspLifecycleManager) + mock_lifecycle.health_check = MagicMock(return_value=True) + mock_lifecycle.get_client = MagicMock(return_value=mock_client) + + runtime = LspRuntime(lifecycle_manager=mock_lifecycle) + # Do NOT call start_server — no workspace path registered + + context.lspc_runtime = runtime + context.lspc_mock_client = mock_client + context.lspc_error = None + + +# --------------------------------------------------------------------------- +# When steps +# --------------------------------------------------------------------------- + + +@when("lspc I call read_file with the workspace path") +def step_lspc_read_file_with_workspace(context: Context) -> None: + # Determine which file to use: inside or outside + file_path = getattr(context, "lspc_inside_file", None) or getattr( + context, "lspc_outside_file", None + ) + try: + context.lspc_file_content = LspRuntime._read_file( + file_path, context.lspc_workspace + ) + context.lspc_error = None + except Exception as exc: + context.lspc_error = exc + context.lspc_file_content = None + + +@when("lspc I call read_file with a traversal path and the workspace path") +def step_lspc_read_file_traversal(context: Context) -> None: + # Build a traversal path: workspace/subdir/../../outside_file + outside_file = context.lspc_outside_file + workspace = context.lspc_workspace + # Construct a path that starts inside the workspace but traverses out + traversal_path = os.path.join(workspace, "subdir", "..", "..", outside_file.lstrip("/")) + try: + context.lspc_file_content = LspRuntime._read_file( + traversal_path, workspace + ) + context.lspc_error = None + except Exception as exc: + context.lspc_error = exc + context.lspc_file_content = None + + +@when("lspc I call read_file without a workspace path") +def step_lspc_read_file_no_workspace(context: Context) -> None: + file_path = context.lspc_outside_file + try: + context.lspc_file_content = LspRuntime._read_file(file_path) + context.lspc_error = None + except Exception as exc: + context.lspc_error = exc + context.lspc_file_content = None + + +@when('lspc I try to get diagnostics for "{name}" on the outside file') +def step_lspc_get_diagnostics_outside(context: Context, name: str) -> None: + try: + context.lspc_result = context.lspc_runtime.get_diagnostics( + name, context.lspc_outside_file + ) + context.lspc_error = None + except Exception as exc: + context.lspc_error = exc + context.lspc_result = None + + +@when('lspc I get diagnostics for "{name}" on the inside file') +def step_lspc_get_diagnostics_inside(context: Context, name: str) -> None: + try: + context.lspc_result = context.lspc_runtime.get_diagnostics( + name, context.lspc_inside_file + ) + context.lspc_error = None + except Exception as exc: + context.lspc_error = exc + context.lspc_result = None + + +@when('lspc I get diagnostics for "{name}" on the outside file') +def step_lspc_get_diagnostics_outside_no_ws(context: Context, name: str) -> None: + try: + context.lspc_result = context.lspc_runtime.get_diagnostics( + name, context.lspc_outside_file + ) + context.lspc_error = None + except Exception as exc: + context.lspc_error = exc + context.lspc_result = None + + +@when( + 'lspc I try to get completions for "{name}" on the outside file' + " at line {line:d} column {col:d}" +) +def step_lspc_get_completions_outside( + context: Context, name: str, line: int, col: int +) -> None: + try: + context.lspc_result = context.lspc_runtime.get_completions( + name, context.lspc_outside_file, line, col + ) + context.lspc_error = None + except Exception as exc: + context.lspc_error = exc + context.lspc_result = None + + +@when( + 'lspc I try to get hover for "{name}" on the outside file' + " at line {line:d} column {col:d}" +) +def step_lspc_get_hover_outside( + context: Context, name: str, line: int, col: int +) -> None: + try: + context.lspc_result = context.lspc_runtime.get_hover( + name, context.lspc_outside_file, line, col + ) + context.lspc_error = None + except Exception as exc: + context.lspc_error = exc + context.lspc_result = None + + +@when( + 'lspc I try to get definitions for "{name}" on the outside file' + " at line {line:d} column {col:d}" +) +def step_lspc_get_definitions_outside( + context: Context, name: str, line: int, col: int +) -> None: + try: + context.lspc_result = context.lspc_runtime.get_definitions( + name, context.lspc_outside_file, line, col + ) + context.lspc_error = None + except Exception as exc: + context.lspc_error = exc + context.lspc_result = None + + +# --------------------------------------------------------------------------- +# Then steps +# --------------------------------------------------------------------------- + + +@then('lspc the file content should be "{expected}"') +def step_lspc_file_content(context: Context, expected: str) -> None: + assert context.lspc_error is None, ( + f"Expected no error, got {context.lspc_error}" + ) + assert context.lspc_file_content == expected, ( + f"Expected '{expected}', got '{context.lspc_file_content}'" + ) + + +@then("lspc no error should be raised") +def step_lspc_no_error(context: Context) -> None: + assert context.lspc_error is None, ( + f"Expected no error, got {type(context.lspc_error).__name__}: " + f"{context.lspc_error}" + ) + + +@then('lspc an LspError should be raised with message containing "{msg}"') +def step_lspc_lsp_error_msg(context: Context, msg: str) -> None: + assert context.lspc_error is not None, ( + "Expected an LspError but no error occurred" + ) + assert isinstance(context.lspc_error, LspError), ( + f"Expected LspError, got {type(context.lspc_error).__name__}: " + f"{context.lspc_error}" + ) + assert msg in str(context.lspc_error), ( + f"Expected '{msg}' in error message, got: {context.lspc_error}" + ) + + +@then("lspc diagnostics should be returned as a list") +def step_lspc_diagnostics_is_list(context: Context) -> None: + assert context.lspc_error is None, ( + f"Expected no error, got {context.lspc_error}" + ) + assert isinstance(context.lspc_result, list), ( + f"Expected list, got {type(context.lspc_result)}" + ) diff --git a/src/cleveragents/lsp/runtime.py b/src/cleveragents/lsp/runtime.py index cb696a498..cdb37f3fd 100644 --- a/src/cleveragents/lsp/runtime.py +++ b/src/cleveragents/lsp/runtime.py @@ -55,6 +55,8 @@ class LspRuntime: ) -> None: self._registry = registry or LspRegistry() self._lifecycle = lifecycle_manager or LspLifecycleManager() + # Maps server name to resolved workspace root for path containment checks. + self._workspace_paths: dict[str, str] = {} @property def registry(self) -> LspRegistry: @@ -98,6 +100,7 @@ class LspRuntime: ) self._lifecycle.start_server(config, workspace_path) + self._workspace_paths[name] = os.path.realpath(workspace_path) def stop_server(self, name: str) -> None: """Stop the LSP server identified by *name*. @@ -117,6 +120,7 @@ class LspRuntime: logger.info("lsp.runtime.stopping_server", server=name) self._lifecycle.stop_server(name) + self._workspace_paths.pop(name, None) def get_diagnostics(self, name: str, file_path: str) -> list[Any]: """Retrieve diagnostics for *file_path* from the named server. @@ -135,7 +139,8 @@ class LspRuntime: Raises: ValueError: If *name* or *file_path* is empty. LspServerNotFoundError: If the server is not running. - LspError: If the server has crashed. + LspError: If the server has crashed or the file is outside + the workspace. """ if not name: raise ValueError("name must be a non-empty string") @@ -146,8 +151,9 @@ class LspRuntime: uri = self._path_to_uri(file_path) # Open the file so the server analyses it + workspace_path = self._workspace_paths.get(name) try: - text = self._read_file(file_path) + text = self._read_file(file_path, workspace_path) except OSError as exc: raise LspError( f"Cannot read file for diagnostics: {file_path}", @@ -192,7 +198,8 @@ class LspRuntime: Raises: ValueError: If inputs are invalid. LspServerNotFoundError: If the server is not running. - LspError: If the server has crashed. + LspError: If the server has crashed or the file is outside + the workspace. """ if not name: raise ValueError("name must be a non-empty string") @@ -207,8 +214,9 @@ class LspRuntime: uri = self._path_to_uri(file_path) # Open the file + workspace_path = self._workspace_paths.get(name) try: - text = self._read_file(file_path) + text = self._read_file(file_path, workspace_path) except OSError as exc: raise LspError( f"Cannot read file for completions: {file_path}", @@ -249,6 +257,9 @@ class LspRuntime: Returns: Hover result dict or ``None`` if no info is available. + + Raises: + LspError: If the file is outside the workspace. """ if not name: raise ValueError("name must be a non-empty string") @@ -262,8 +273,9 @@ class LspRuntime: client = self._get_healthy_client(name) uri = self._path_to_uri(file_path) + workspace_path = self._workspace_paths.get(name) try: - text = self._read_file(file_path) + text = self._read_file(file_path, workspace_path) except OSError as exc: raise LspError( f"Cannot read file for hover: {file_path}", @@ -303,6 +315,9 @@ class LspRuntime: Returns: List of Location dicts. + + Raises: + LspError: If the file is outside the workspace. """ if not name: raise ValueError("name must be a non-empty string") @@ -316,8 +331,9 @@ class LspRuntime: client = self._get_healthy_client(name) uri = self._path_to_uri(file_path) + workspace_path = self._workspace_paths.get(name) try: - text = self._read_file(file_path) + text = self._read_file(file_path, workspace_path) except OSError as exc: raise LspError( f"Cannot read file for definitions: {file_path}", @@ -370,13 +386,44 @@ class LspRuntime: return f"file://{file_path}" @staticmethod - def _read_file(file_path: str) -> str: + def _read_file(file_path: str, workspace_path: str | None = None) -> str: """Read file contents as UTF-8 text. + When *workspace_path* is provided the resolved file path must be + contained within the workspace directory. This prevents path + traversal attacks where a caller supplies a path such as + ``../../etc/passwd`` to escape the workspace root. + + Args: + file_path: Path to the file to read. + workspace_path: Optional workspace root. When supplied, the + resolved *file_path* must start with the resolved + *workspace_path* or an :class:`LspError` is raised. + Raises: - LspError: If the resolved path is a directory or device. + LspError: If the resolved path is a directory, device, or + (when *workspace_path* is given) outside the workspace. """ resolved = os.path.realpath(file_path) + if workspace_path is not None: + resolved_workspace = os.path.realpath(workspace_path) + # Ensure the file is strictly inside the workspace directory. + # We append os.sep so that a workspace of "/tmp/ws" does not + # accidentally match "/tmp/ws2/file.py". + ws_prefix = ( + resolved_workspace + if resolved_workspace.endswith(os.sep) + else resolved_workspace + os.sep + ) + if resolved != resolved_workspace and not resolved.startswith(ws_prefix): + raise LspError( + f"Path traversal detected: '{file_path}' is outside workspace", + details={ + "file": file_path, + "resolved": resolved, + "workspace": workspace_path, + }, + ) if not os.path.isfile(resolved): raise LspError( f"Not a regular file: {file_path}", @@ -448,6 +495,7 @@ class LspRuntime: def stop_all(self) -> None: """Shut down all running LSP servers.""" self._lifecycle.stop_all() + self._workspace_paths.clear() __all__ = [ -- 2.52.0 From 4f23ece138c170cbb526adb1dea7a5b5816afe36 Mon Sep 17 00:00:00 2001 From: CleverThis Date: Sun, 19 Apr 2026 11:31:55 +0000 Subject: [PATCH 2/2] style: apply ruff formatting to lsp_path_containment_steps.py Fix CI lint failure by running ruff format on the step definitions file that was missing auto-formatting before the PR was submitted. --- features/steps/lsp_path_containment_steps.py | 25 +++++++------------- 1 file changed, 9 insertions(+), 16 deletions(-) diff --git a/features/steps/lsp_path_containment_steps.py b/features/steps/lsp_path_containment_steps.py index 7f6e074e0..fce688f22 100644 --- a/features/steps/lsp_path_containment_steps.py +++ b/features/steps/lsp_path_containment_steps.py @@ -48,6 +48,7 @@ def step_lspc_create_workspace(context: Context) -> None: def cleanup() -> None: import shutil + if os.path.exists(workspace): shutil.rmtree(workspace, ignore_errors=True) @@ -87,9 +88,7 @@ def step_lspc_create_outside_file(context: Context) -> None: context.add_cleanup(cleanup) -@given( - 'lspc I create an LspRuntime with a healthy mock server "{name}" and workspace' -) +@given('lspc I create an LspRuntime with a healthy mock server "{name}" and workspace') def step_lspc_create_runtime_with_workspace(context: Context, name: str) -> None: mock_client = _make_mock_client() @@ -161,11 +160,11 @@ def step_lspc_read_file_traversal(context: Context) -> None: outside_file = context.lspc_outside_file workspace = context.lspc_workspace # Construct a path that starts inside the workspace but traverses out - traversal_path = os.path.join(workspace, "subdir", "..", "..", outside_file.lstrip("/")) + traversal_path = os.path.join( + workspace, "subdir", "..", "..", outside_file.lstrip("/") + ) try: - context.lspc_file_content = LspRuntime._read_file( - traversal_path, workspace - ) + context.lspc_file_content = LspRuntime._read_file(traversal_path, workspace) context.lspc_error = None except Exception as exc: context.lspc_error = exc @@ -277,9 +276,7 @@ def step_lspc_get_definitions_outside( @then('lspc the file content should be "{expected}"') def step_lspc_file_content(context: Context, expected: str) -> None: - assert context.lspc_error is None, ( - f"Expected no error, got {context.lspc_error}" - ) + assert context.lspc_error is None, f"Expected no error, got {context.lspc_error}" assert context.lspc_file_content == expected, ( f"Expected '{expected}', got '{context.lspc_file_content}'" ) @@ -295,9 +292,7 @@ def step_lspc_no_error(context: Context) -> None: @then('lspc an LspError should be raised with message containing "{msg}"') def step_lspc_lsp_error_msg(context: Context, msg: str) -> None: - assert context.lspc_error is not None, ( - "Expected an LspError but no error occurred" - ) + assert context.lspc_error is not None, "Expected an LspError but no error occurred" assert isinstance(context.lspc_error, LspError), ( f"Expected LspError, got {type(context.lspc_error).__name__}: " f"{context.lspc_error}" @@ -309,9 +304,7 @@ def step_lspc_lsp_error_msg(context: Context, msg: str) -> None: @then("lspc diagnostics should be returned as a list") def step_lspc_diagnostics_is_list(context: Context) -> None: - assert context.lspc_error is None, ( - f"Expected no error, got {context.lspc_error}" - ) + assert context.lspc_error is None, f"Expected no error, got {context.lspc_error}" assert isinstance(context.lspc_result, list), ( f"Expected list, got {type(context.lspc_result)}" ) -- 2.52.0