fix(lsp): validate workspace boundary in _read_file to prevent path traversal #10644

Merged
HAL9000 merged 4 commits from fix/v360/lsp-path-traversal-file-reading into master 2026-06-06 04:21:50 +00:00
3 changed files with 525 additions and 47 deletions
@@ -0,0 +1,87 @@
Feature: LSP runtime path traversal security
Outdated
Review

BLOCKING: Missing companion TDD issue. Per CONTRIBUTING.md bug fix workflow, a Type/Testing TDD issue with a failing regression test must be created BEFORE the fix PR. Issue #7215 subtask requires: “Create companion Type/Testing TDD issue (per Bug Fix Workflow in CONTRIBUTING.md) before implementing the fix.” No TDD issue is referenced in the PR body.

BLOCKING: Missing companion TDD issue. Per CONTRIBUTING.md bug fix workflow, a Type/Testing TDD issue with a failing regression test must be created BEFORE the fix PR. Issue #7215 subtask requires: “Create companion Type/Testing TDD issue (per Bug Fix Workflow in CONTRIBUTING.md) before implementing the fix.” No TDD issue is referenced in the PR body.
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"
@@ -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)}"
)
+72 -47
View File
1
@@ -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:
1
@@ -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}",
1
@@ -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__ = [