fix(lsp): validate workspace boundary in _read_file to prevent path traversal #10644
@@ -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"
|
||||
@@ -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)}"
|
||||
)
|
||||
@@ -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__ = [
|
||||
|
||||
Reference in New Issue
Block a user
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.