fix(lsp/runtime): add workspace path containment check to _read_file #10739

Merged
HAL9000 merged 2 commits from fix/issue-10490-lsp-path-containment into master 2026-04-23 09:38:33 +00:00
3 changed files with 449 additions and 8 deletions
+83
View File
@@ -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
@@ -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)}"
)
+56 -8
View File
@@ -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__ = [