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..fce688f22 --- /dev/null +++ b/features/steps/lsp_path_containment_steps.py @@ -0,0 +1,310 @@ +"""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__ = [