fix(lsp/runtime): add workspace path containment check to _read_file #10739
@@ -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)}"
|
||||
)
|
||||
@@ -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__ = [
|
||||
|
||||
Reference in New Issue
Block a user