diff --git a/features/lsp_path_traversal_security.feature b/features/lsp_path_traversal_security.feature new file mode 100644 index 000000000..5d8d904fb --- /dev/null +++ b/features/lsp_path_traversal_security.feature @@ -0,0 +1,87 @@ +Feature: LSP runtime path traversal security + As a security-conscious platform operator + I want the LSP runtime to reject file paths that escape the workspace + So that malicious LSP clients cannot read sensitive system files + + # ── _validate_workspace_path unit tests ────────────────────────── + + Scenario: validate_workspace_path accepts a file directly inside workspace + Given lspsec a workspace root at a temp directory + And lspsec a file "safe.py" inside the workspace + When lspsec I call validate_workspace_path with the file inside workspace + Then lspsec no error should be raised + + Scenario: validate_workspace_path accepts a file in a subdirectory of workspace + Given lspsec a workspace root at a temp directory + And lspsec a file "subdir/nested.py" inside the workspace + When lspsec I call validate_workspace_path with the nested file + Then lspsec no error should be raised + + Scenario: validate_workspace_path rejects a path outside the workspace + Given lspsec a workspace root at a temp directory + When lspsec I call validate_workspace_path with absolute path "/etc/passwd" + Then lspsec an LspError should be raised with message containing "Path traversal attempt blocked" + + Scenario: validate_workspace_path rejects a path that is a sibling directory + Given lspsec a workspace root at a temp directory + And lspsec a sibling directory exists next to the workspace + When lspsec I call validate_workspace_path with a file in the sibling directory + Then lspsec an LspError should be raised with message containing "Path traversal attempt blocked" + + Scenario: validate_workspace_path rejects the workspace root itself + Given lspsec a workspace root at a temp directory + When lspsec I call validate_workspace_path with the workspace root itself + Then lspsec an LspError should be raised with message containing "Path traversal attempt blocked" + + # ── _read_file with workspace_root ─────────────────────────────── + + Scenario: read_file succeeds for a valid file inside workspace + Given lspsec a workspace root at a temp directory + And lspsec a file "hello.py" inside the workspace with content "x = 1" + When lspsec I call read_file on workspace file "hello.py" with the workspace root + Then lspsec the file content should be "x = 1" + + Scenario: read_file blocks path traversal via dot-dot segments + Given lspsec a workspace root at a temp directory + And lspsec a file "hello.py" inside the workspace with content "x = 1" + When lspsec I call read_file on a dot-dot traversal path targeting etc-passwd + Then lspsec an LspError should be raised with message containing "Path traversal attempt blocked" + + Scenario: read_file blocks an absolute path outside the workspace + Given lspsec a workspace root at a temp directory + When lspsec I call read_file on absolute path "/etc/hostname" with the workspace root + Then lspsec an LspError should be raised with message containing "Path traversal attempt blocked" + + Scenario: read_file blocks a symlink that points outside the workspace + Given lspsec a workspace root at a temp directory + And lspsec a symlink "evil_link.py" inside the workspace pointing to "/etc/passwd" + When lspsec I call read_file on the symlink with the workspace root + Then lspsec an LspError should be raised with message containing "Path traversal attempt blocked" + + Scenario: read_file without workspace_root skips boundary check + Given lspsec a temp file outside any workspace with content "secret" + When lspsec I call read_file on the temp file without workspace_root + Then lspsec the file content should be "secret" + + # ── LspRuntime.get_diagnostics with workspace boundary ─────────── + + Scenario: get_diagnostics blocks path traversal when workspace is registered + Given lspsec an LspRuntime with workspace root at a temp directory + And lspsec a healthy mock server "local/pyright" registered for the workspace + When lspsec I try to get diagnostics for "local/pyright" on absolute path "/etc/passwd" + Then lspsec an LspError should be raised with message containing "Path traversal attempt blocked" + + Scenario: get_diagnostics succeeds for a valid file inside the workspace + Given lspsec an LspRuntime with workspace root at a temp directory + And lspsec a healthy mock server "local/pyright" registered for the workspace + And lspsec a file "main.py" inside the workspace with content "print('hello')" + When lspsec I get diagnostics for "local/pyright" on workspace file "main.py" + Then lspsec diagnostics should be returned as a list + + # ── LspRuntime.get_completions with workspace boundary ─────────── + + Scenario: get_completions blocks path traversal when workspace is registered + Given lspsec an LspRuntime with workspace root at a temp directory + And lspsec a healthy mock server "local/pyright" registered for the workspace + When lspsec I try to get completions for "local/pyright" on absolute path "/etc/passwd" at line 1 column 1 + Then lspsec an LspError should be raised with message containing "Path traversal attempt blocked" diff --git a/features/steps/lsp_path_traversal_security_steps.py b/features/steps/lsp_path_traversal_security_steps.py new file mode 100644 index 000000000..1bbadbeaf --- /dev/null +++ b/features/steps/lsp_path_traversal_security_steps.py @@ -0,0 +1,366 @@ +"""Step definitions for lsp_path_traversal_security.feature. + +Tests the path traversal vulnerability fix in LspRuntime._read_file +and the new _validate_workspace_path helper. + +Uses the ``lspsec`` 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.models import LspServerConfig +from cleveragents.lsp.registry import LspRegistry +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=[]) + return client + + +def _make_config(name: str = "local/pyright") -> LspServerConfig: + """Build a minimal server config for testing.""" + return LspServerConfig( + name=name, + command="echo", + languages=["python"], + ) + + +# --------------------------------------------------------------------------- +# Given steps +# --------------------------------------------------------------------------- + + +@given("lspsec a workspace root at a temp directory") +def step_lspsec_workspace_root(context: Context) -> None: + tmpdir = tempfile.mkdtemp(prefix="lspsec_ws_") + context.lspsec_workspace = tmpdir + context.lspsec_error = None + + def cleanup() -> None: + import shutil + + shutil.rmtree(tmpdir, ignore_errors=True) + + context.add_cleanup(cleanup) + + +@given('lspsec a file "{filename}" inside the workspace') +def step_lspsec_file_in_workspace(context: Context, filename: str) -> None: + filepath = os.path.join(context.lspsec_workspace, filename) + os.makedirs(os.path.dirname(filepath), exist_ok=True) + with open(filepath, "w", encoding="utf-8") as f: + f.write("") + context.lspsec_file_in_workspace = filepath + + +@given('lspsec a file "{filename}" inside the workspace with content "{content}"') +def step_lspsec_file_in_workspace_with_content( + context: Context, filename: str, content: str +) -> None: + filepath = os.path.join(context.lspsec_workspace, filename) + os.makedirs(os.path.dirname(filepath), exist_ok=True) + with open(filepath, "w", encoding="utf-8") as f: + f.write(content) + context.lspsec_file_in_workspace = filepath + + +@given("lspsec a sibling directory exists next to the workspace") +def step_lspsec_sibling_dir(context: Context) -> None: + sibling = tempfile.mkdtemp(prefix="lspsec_sibling_") + context.lspsec_sibling_dir = sibling + sibling_file = os.path.join(sibling, "secret.txt") + with open(sibling_file, "w", encoding="utf-8") as f: + f.write("sibling secret") + context.lspsec_sibling_file = sibling_file + + def cleanup() -> None: + import shutil + + shutil.rmtree(sibling, ignore_errors=True) + + context.add_cleanup(cleanup) + + +@given('lspsec a symlink "{linkname}" inside the workspace pointing to "{target}"') +def step_lspsec_symlink_in_workspace( + context: Context, linkname: str, target: str +) -> None: + link_path = os.path.join(context.lspsec_workspace, linkname) + os.symlink(target, link_path) + context.lspsec_symlink_path = link_path + + def cleanup() -> None: + if os.path.lexists(link_path): + os.unlink(link_path) + + context.add_cleanup(cleanup) + + +@given('lspsec a temp file outside any workspace with content "{content}"') +def step_lspsec_temp_file_outside_workspace(context: Context, content: str) -> None: + fd, path = tempfile.mkstemp(prefix="lspsec_outside_", suffix=".txt") + with os.fdopen(fd, "w") as f: + f.write(content) + context.lspsec_outside_file = path + + def cleanup() -> None: + if os.path.exists(path): + os.unlink(path) + + context.add_cleanup(cleanup) + + +@given("lspsec an LspRuntime with workspace root at a temp directory") +def step_lspsec_runtime_with_workspace(context: Context) -> None: + tmpdir = tempfile.mkdtemp(prefix="lspsec_rt_ws_") + context.lspsec_workspace = tmpdir + context.lspsec_error = None + + def cleanup() -> None: + import shutil + + shutil.rmtree(tmpdir, ignore_errors=True) + + context.add_cleanup(cleanup) + + # Create runtime with a mock lifecycle (no real server needed) + mock_lifecycle = MagicMock(spec=LspLifecycleManager) + mock_lifecycle.start_server = MagicMock() + mock_lifecycle.stop_server = MagicMock() + mock_lifecycle.stop_all = MagicMock() + mock_lifecycle.health_check = MagicMock(return_value=True) + mock_lifecycle.get_client = MagicMock(return_value=_make_mock_client()) + + context.lspsec_runtime = LspRuntime(lifecycle_manager=mock_lifecycle) + context.lspsec_mock_lifecycle = mock_lifecycle + + +@given('lspsec a healthy mock server "{name}" registered for the workspace') +def step_lspsec_register_server_for_workspace(context: Context, name: str) -> None: + registry = LspRegistry() + config = _make_config(name) + registry.register(config) + context.lspsec_runtime._registry = registry + + mock_client = _make_mock_client() + context.lspsec_mock_lifecycle.get_client = MagicMock(return_value=mock_client) + context.lspsec_mock_client = mock_client + + # Simulate that start_server was called so workspace root is stored + context.lspsec_runtime._workspace_roots[name] = context.lspsec_workspace + context.lspsec_server_name = name + + +# --------------------------------------------------------------------------- +# When steps +# --------------------------------------------------------------------------- + + +@when("lspsec I call validate_workspace_path with the file inside workspace") +def step_lspsec_validate_file_inside(context: Context) -> None: + resolved = os.path.realpath(context.lspsec_file_in_workspace) + try: + LspRuntime._validate_workspace_path(resolved, context.lspsec_workspace) + context.lspsec_error = None + except LspError as exc: + context.lspsec_error = exc + + +@when("lspsec I call validate_workspace_path with the nested file") +def step_lspsec_validate_nested_file(context: Context) -> None: + resolved = os.path.realpath(context.lspsec_file_in_workspace) + try: + LspRuntime._validate_workspace_path(resolved, context.lspsec_workspace) + context.lspsec_error = None + except LspError as exc: + context.lspsec_error = exc + + +@when('lspsec I call validate_workspace_path with absolute path "{path}"') +def step_lspsec_validate_absolute_path(context: Context, path: str) -> None: + resolved = os.path.realpath(path) + try: + LspRuntime._validate_workspace_path(resolved, context.lspsec_workspace) + context.lspsec_error = None + except LspError as exc: + context.lspsec_error = exc + + +@when("lspsec I call validate_workspace_path with a file in the sibling directory") +def step_lspsec_validate_sibling_file(context: Context) -> None: + resolved = os.path.realpath(context.lspsec_sibling_file) + try: + LspRuntime._validate_workspace_path(resolved, context.lspsec_workspace) + context.lspsec_error = None + except LspError as exc: + context.lspsec_error = exc + + +@when("lspsec I call validate_workspace_path with the workspace root itself") +def step_lspsec_validate_workspace_root_itself(context: Context) -> None: + resolved = os.path.realpath(context.lspsec_workspace) + try: + LspRuntime._validate_workspace_path(resolved, context.lspsec_workspace) + context.lspsec_error = None + except LspError as exc: + context.lspsec_error = exc + + +@when('lspsec I call read_file on workspace file "{filename}" with the workspace root') +def step_lspsec_read_workspace_file(context: Context, filename: str) -> None: + filepath = os.path.join(context.lspsec_workspace, filename) + try: + context.lspsec_result = LspRuntime._read_file( + filepath, workspace_root=context.lspsec_workspace + ) + context.lspsec_error = None + except LspError as exc: + context.lspsec_error = exc + + +@when("lspsec I call read_file on a dot-dot traversal path targeting etc-passwd") +def step_lspsec_read_file_dotdot(context: Context) -> None: + # Build a path like /tmp/lspsec_ws_xxx/../../etc/passwd + traversal = os.path.join(context.lspsec_workspace, "..", "..", "etc", "passwd") + try: + context.lspsec_result = LspRuntime._read_file( + traversal, workspace_root=context.lspsec_workspace + ) + context.lspsec_error = None + except LspError as exc: + context.lspsec_error = exc + + +@when('lspsec I call read_file on absolute path "{path}" with the workspace root') +def step_lspsec_read_file_absolute_path(context: Context, path: str) -> None: + try: + context.lspsec_result = LspRuntime._read_file( + path, workspace_root=context.lspsec_workspace + ) + context.lspsec_error = None + except LspError as exc: + context.lspsec_error = exc + + +@when("lspsec I call read_file on the symlink with the workspace root") +def step_lspsec_read_file_symlink(context: Context) -> None: + try: + context.lspsec_result = LspRuntime._read_file( + context.lspsec_symlink_path, + workspace_root=context.lspsec_workspace, + ) + context.lspsec_error = None + except LspError as exc: + context.lspsec_error = exc + + +@when("lspsec I call read_file on the temp file without workspace_root") +def step_lspsec_read_file_no_workspace(context: Context) -> None: + try: + context.lspsec_result = LspRuntime._read_file(context.lspsec_outside_file) + context.lspsec_error = None + except LspError as exc: + context.lspsec_error = exc + + +@when('lspsec I try to get diagnostics for "{name}" on absolute path "{path}"') +def step_lspsec_get_diagnostics_traversal( + context: Context, name: str, path: str +) -> None: + try: + context.lspsec_runtime.get_diagnostics(name, path) + context.lspsec_error = None + except LspError as exc: + context.lspsec_error = exc + + +@when('lspsec I get diagnostics for "{name}" on workspace file "{filename}"') +def step_lspsec_get_diagnostics_valid( + context: Context, name: str, filename: str +) -> None: + filepath = os.path.join(context.lspsec_workspace, filename) + try: + context.lspsec_result = context.lspsec_runtime.get_diagnostics(name, filepath) + context.lspsec_error = None + except LspError as exc: + context.lspsec_error = exc + + +@when( + 'lspsec I try to get completions for "{name}" on absolute path "{path}"' + " at line {line:d} column {col:d}" +) +def step_lspsec_get_completions_traversal( + context: Context, name: str, path: str, line: int, col: int +) -> None: + try: + context.lspsec_runtime.get_completions(name, path, line, col) + context.lspsec_error = None + except LspError as exc: + context.lspsec_error = exc + + +# --------------------------------------------------------------------------- +# Then steps +# --------------------------------------------------------------------------- + + +@then("lspsec no error should be raised") +def step_lspsec_no_error(context: Context) -> None: + assert context.lspsec_error is None, ( + f"Expected no error but got: {context.lspsec_error}" + ) + + +@then('lspsec an LspError should be raised with message containing "{msg}"') +def step_lspsec_lsp_error_msg(context: Context, msg: str) -> None: + assert context.lspsec_error is not None, ( + "Expected an LspError but no error was raised" + ) + assert isinstance(context.lspsec_error, LspError), ( + f"Expected LspError, got {type(context.lspsec_error).__name__}: " + f"{context.lspsec_error}" + ) + assert msg in str(context.lspsec_error), ( + f"Expected '{msg}' in error message, got: {context.lspsec_error}" + ) + + +@then('lspsec the file content should be "{expected}"') +def step_lspsec_file_content(context: Context, expected: str) -> None: + assert context.lspsec_error is None, ( + f"Expected no error but got: {context.lspsec_error}" + ) + assert context.lspsec_result == expected, ( + f"Expected content '{expected}', got '{context.lspsec_result}'" + ) + + +@then("lspsec diagnostics should be returned as a list") +def step_lspsec_diagnostics_list(context: Context) -> None: + assert context.lspsec_error is None, ( + f"Expected no error but got: {context.lspsec_error}" + ) + assert isinstance(context.lspsec_result, list), ( + f"Expected list, got {type(context.lspsec_result)}" + ) diff --git a/src/cleveragents/lsp/runtime.py b/src/cleveragents/lsp/runtime.py index 4677a50aa..56bc99969 100644 --- a/src/cleveragents/lsp/runtime.py +++ b/src/cleveragents/lsp/runtime.py @@ -63,8 +63,8 @@ class LspRuntime: self._lifecycle = lifecycle_manager else: self._lifecycle = LspLifecycleManager() - # Maps server name to resolved workspace root for path containment checks. - self._workspace_paths: dict[str, str] = {} + # Maps server name to its workspace root for path containment checks. + self._workspace_roots: dict[str, str] = {} @property def registry(self) -> LspRegistry: @@ -108,7 +108,7 @@ class LspRuntime: ) self._lifecycle.start_server(config, workspace_path) - self._workspace_paths[name] = os.path.realpath(workspace_path) + self._workspace_roots[name] = workspace_path def stop_server(self, name: str) -> None: """Stop the LSP server identified by *name*. @@ -128,7 +128,7 @@ class LspRuntime: logger.info("lsp.runtime.stopping_server", server=name) self._lifecycle.stop_server(name) - self._workspace_paths.pop(name, None) + self._workspace_roots.pop(name, None) def get_diagnostics(self, name: str, file_path: str) -> list[Any]: """Retrieve diagnostics for *file_path* from the named server. @@ -147,8 +147,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 or the file is outside - the workspace. + LspError: If the server has crashed or path traversal is + detected. """ if not name: raise ValueError("name must be a non-empty string") @@ -159,9 +159,11 @@ 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, workspace_path) + text = self._read_file( + file_path, + workspace_root=self._workspace_roots.get(name), + ) except OSError as exc: raise LspError( f"Cannot read file for diagnostics: {file_path}", @@ -206,8 +208,8 @@ class LspRuntime: Raises: ValueError: If inputs are invalid. LspServerNotFoundError: If the server is not running. - LspError: If the server has crashed or the file is outside - the workspace. + LspError: If the server has crashed or path traversal is + detected. """ if not name: raise ValueError("name must be a non-empty string") @@ -222,9 +224,11 @@ 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, workspace_path) + text = self._read_file( + file_path, + workspace_root=self._workspace_roots.get(name), + ) except OSError as exc: raise LspError( f"Cannot read file for completions: {file_path}", @@ -281,9 +285,11 @@ 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, workspace_path) + text = self._read_file( + file_path, + workspace_root=self._workspace_roots.get(name), + ) except OSError as exc: raise LspError( f"Cannot read file for hover: {file_path}", @@ -339,9 +345,11 @@ 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, workspace_path) + text = self._read_file( + file_path, + workspace_root=self._workspace_roots.get(name), + ) except OSError as exc: raise LspError( f"Cannot read file for definitions: {file_path}", @@ -394,44 +402,61 @@ class LspRuntime: return f"file://{file_path}" @staticmethod - def _read_file(file_path: str, workspace_path: str | None = None) -> str: - """Read file contents as UTF-8 text. + def _validate_workspace_path(resolved: str, workspace_root: str) -> None: + """Validate that *resolved* is contained within *workspace_root*. - 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. + Prevents path traversal attacks by ensuring the canonicalised + file path stays strictly inside the workspace boundary. 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. + resolved: The ``os.path.realpath``-resolved absolute path. + workspace_root: The workspace root directory (will be + canonicalised internally). + + Raises: + LspError: If *resolved* does not start with the canonicalised + workspace root, indicating a path traversal attempt. + """ + canonical_root = os.path.realpath(workspace_root) + # Ensure the root ends with a separator so that a directory whose + # name is a prefix of another directory is not falsely accepted + # (e.g. /workspace vs /workspace-evil). + root_prefix = canonical_root + os.sep + if not resolved.startswith(root_prefix): + raise LspError( + f"Path traversal attempt blocked: '{resolved}' is outside " + f"workspace '{canonical_root}'", + details={"resolved": resolved, "workspace_root": canonical_root}, + ) + + @staticmethod + def _read_file( + file_path: str, + workspace_root: str | None = None, + ) -> str: + """Read file contents as UTF-8 text. + + When *workspace_root* is provided the resolved path is validated + against the workspace boundary before the file is opened. This + prevents path traversal attacks where a malicious LSP client + supplies a payload such as ``../../../etc/passwd``. + + Args: + file_path: Path to the file (may be relative or contain + ``..`` components). + workspace_root: Optional workspace root directory. When + supplied, the resolved path must be a descendant of this + directory; otherwise :class:`LspError` is raised. Raises: LspError: If the resolved path is a directory, device, or - (when *workspace_path* is given) outside the workspace. + (when *workspace_root* 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 workspace_root is not None: + LspRuntime._validate_workspace_path(resolved, workspace_root) + if not os.path.isfile(resolved): raise LspError( f"Not a regular file: {file_path}", @@ -503,7 +528,7 @@ class LspRuntime: def stop_all(self) -> None: """Shut down all running LSP servers.""" self._lifecycle.stop_all() - self._workspace_paths.clear() + self._workspace_roots.clear() __all__ = [