From 9ec0a3d425c3b2e8faa87529d15034403d219b05 Mon Sep 17 00:00:00 2001 From: HAL9000 Date: Sat, 18 Apr 2026 20:52:39 +0000 Subject: [PATCH 1/4] fix(lsp): validate environment variables in StdioTransport to prevent code injection --- features/lsp_transport_coverage.feature | 17 ++++ .../steps/lsp_transport_coverage_steps.py | 63 +++++++++++++- src/cleveragents/lsp/transport.py | 87 +++++++++++++++++-- 3 files changed, 159 insertions(+), 8 deletions(-) diff --git a/features/lsp_transport_coverage.feature b/features/lsp_transport_coverage.feature index 07e2186c6..28a9dcc94 100644 --- a/features/lsp_transport_coverage.feature +++ b/features/lsp_transport_coverage.feature @@ -3,6 +3,23 @@ Feature: LSP StdioTransport coverage I need thorough tests for StdioTransport So that all error paths and normal flows are exercised + # ── start() environment validation ────────────────────────────── + + Scenario: ltcov start merges sanitized environment variables + Given ltcov I create a StdioTransport for command "cat" with env {"LSP_LOG_LEVEL":"debug","CUSTOM_PORT":4040} + And ltcov Popen is patched to capture env + When ltcov I try to start the transport + Then ltcov no error should have occurred + And ltcov the captured env should include "LSP_LOG_LEVEL" with value "debug" + And ltcov the captured env should include "CUSTOM_PORT" with value "4040" + + Scenario: ltcov start rejects disallowed environment variables + Given ltcov I create a StdioTransport for command "cat" with env {"LD_PRELOAD":"/tmp/libmalicious.so"} + And ltcov Popen is patched to capture env + When ltcov I try to start the transport + Then ltcov the error should be an LspError with message "Disallowed environment variable" + And ltcov Popen should not have been invoked + # ── start() error paths ───────────────────────────────────────── @tdd_issue @tdd_issue_7044 diff --git a/features/steps/lsp_transport_coverage_steps.py b/features/steps/lsp_transport_coverage_steps.py index 8d94467a8..6c635234d 100644 --- a/features/steps/lsp_transport_coverage_steps.py +++ b/features/steps/lsp_transport_coverage_steps.py @@ -12,6 +12,7 @@ from __future__ import annotations import json import subprocess +from typing import Any from unittest.mock import MagicMock, PropertyMock, patch from behave import given, then, when @@ -32,6 +33,20 @@ def step_ltcov_create_transport(context: Context, cmd: str) -> None: context.ltcov_transport = StdioTransport(command=cmd) context.ltcov_error = None context.ltcov_result = None + context.ltcov_captured_env = None + context.ltcov_popen_call_count = 0 + + +@given('ltcov I create a StdioTransport for command "{cmd}" with env {env_json}') +def step_ltcov_create_transport_with_env( + context: Context, cmd: str, env_json: str +) -> None: + env = json.loads(env_json) + context.ltcov_transport = StdioTransport(command=cmd, env=env) + context.ltcov_error = None + context.ltcov_result = None + context.ltcov_captured_env = None + context.ltcov_popen_call_count = 0 @given("ltcov the transport has a running mock process") @@ -60,6 +75,26 @@ def step_ltcov_popen_oserror(context: Context, msg: str) -> None: context.add_cleanup(patcher.stop) +@given("ltcov Popen is patched to capture env") +def step_ltcov_popen_capture_env(context: Context) -> None: + context.ltcov_captured_env = None + context.ltcov_popen_call_count = 0 + + def fake_popen(cmd: Any, *args: Any, **kwargs: Any) -> MagicMock: + context.ltcov_popen_call_count += 1 + env = kwargs.get("env") + if env is not None: + context.ltcov_captured_env = dict(env) + return _make_mock_process() + + patcher = patch( + "cleveragents.lsp.transport.subprocess.Popen", + side_effect=fake_popen, + ) + patcher.start() + context.add_cleanup(patcher.stop) + + @given("ltcov the transport has a mock process that already exited with code {code:d}") def step_ltcov_exited_process(context: Context, code: int) -> None: proc = _make_mock_process(poll_return=code, returncode=code) @@ -254,7 +289,12 @@ def step_ltcov_body_timeout(context: Context) -> None: # First select call returns ready (for headers), second returns not-ready (body timeout) call_count = [0] - def select_side_effect(rlist, wlist, xlist, timeout=None): + def select_side_effect( + rlist: list[Any], + wlist: list[Any], + xlist: list[Any], + timeout: float | None = None, + ): call_count[0] += 1 if call_count[0] <= 2: # Headers: two readline calls both need select to report ready @@ -405,6 +445,11 @@ def step_ltcov_lsp_error(context: Context, fragment: str) -> None: ) +@then("ltcov no error should have occurred") +def step_ltcov_no_error(context: Context) -> None: + assert context.ltcov_error is None, f"Unexpected error: {context.ltcov_error}" + + @then("ltcov the stop result should be None") def step_ltcov_stop_none(context: Context) -> None: assert context.ltcov_result is None, f"Expected None, got {context.ltcov_result}" @@ -474,3 +519,19 @@ def step_ltcov_value_error(context: Context, fragment: str) -> None: assert fragment in str(context.ltcov_error), ( f"Expected '{fragment}' in '{context.ltcov_error}'" ) + + +@then('ltcov the captured env should include "{key}" with value "{value}"') +def step_ltcov_env_contains(context: Context, key: str, value: str) -> None: + assert context.ltcov_captured_env is not None, "Expected env to be captured" + assert key in context.ltcov_captured_env, f"Key '{key}' missing from env" + assert context.ltcov_captured_env[key] == value, ( + f"Expected '{value}', got '{context.ltcov_captured_env[key]}'" + ) + + +@then("ltcov Popen should not have been invoked") +def step_ltcov_popen_not_called(context: Context) -> None: + assert ( + context.ltcov_popen_call_count == 0 + ), f"Expected no Popen calls, got {context.ltcov_popen_call_count}" diff --git a/src/cleveragents/lsp/transport.py b/src/cleveragents/lsp/transport.py index cc45c672f..9f21bc440 100644 --- a/src/cleveragents/lsp/transport.py +++ b/src/cleveragents/lsp/transport.py @@ -20,6 +20,7 @@ from __future__ import annotations import json import os +import re import select import subprocess import threading @@ -27,6 +28,8 @@ from typing import Any import structlog +from cleveragents.lsp.errors import LspError + logger = structlog.get_logger(__name__) # Maximum time (seconds) to wait for a graceful process termination @@ -41,6 +44,28 @@ _DEFAULT_READ_TIMEOUT = 30.0 # the existing ``server.py`` limit). _MAX_CONTENT_LENGTH = 10 * 1024 * 1024 +# Disallowed environment variables that can hijack dynamic linking +# or otherwise alter interpreter behaviour in unsafe ways. These are +# case-insensitive checks; we normalise to upper-case before matching. +_DISALLOWED_ENV_VARS = { + "LD_PRELOAD", + "LD_LIBRARY_PATH", + "DYLD_INSERT_LIBRARIES", + "DYLD_LIBRARY_PATH", + "PATH", + "PATHEXT", + "PYTHONHOME", + "PYTHONPATH", + "PYTHONSTARTUP", + "PYTHONBREAKPOINT", + "PYTHONUSERBASE", +} + +# Valid environment variable names follow POSIX shell rules. We +# normalise to uppercase but preserve the caller's original casing in +# the merged environment. +_ENV_KEY_PATTERN = re.compile(r"^[A-Z_][A-Z0-9_]*$") + class StdioTransport: """Stdio transport for an LSP server subprocess. @@ -103,7 +128,8 @@ class StdioTransport: if self._process is not None and self.is_alive: raise RuntimeError("Transport already started") - merged_env = {**os.environ, **self._env} + sanitized_env = self._sanitize_env(self._env) + merged_env = {**os.environ, **sanitized_env} cmd = [self._command, *self._args] logger.info( @@ -124,9 +150,6 @@ class StdioTransport: ) except FileNotFoundError as exc: self._process = None - - from cleveragents.lsp.errors import LspError - raise LspError( f"LSP server command not found: {self._command}", details={"command": self._command, "args": self._args}, @@ -139,9 +162,6 @@ class StdioTransport: if self._process is not None: self.stop() self._process = None - - from cleveragents.lsp.errors import LspError - raise LspError( f"Failed to start LSP server: {exc}", details={"command": self._command, "error": str(exc)}, @@ -274,6 +294,59 @@ class StdioTransport: with self._read_lock: return self._read_one_message(effective_timeout) + def _sanitize_env(self, env: dict[str, Any]) -> dict[str, str]: + """Validate and sanitise user-supplied environment variables. + + Rejects variables that can compromise process safety and ensures + all values are strings without null bytes. Raises ``LspError`` + if validation fails. + """ + + sanitized: dict[str, str] = {} + for raw_key, raw_value in env.items(): + if not isinstance(raw_key, str): + raise LspError( + "Environment variable names must be strings", + details={"name": raw_key}, + ) + + key = raw_key.strip() + if not key: + raise LspError( + "Environment variable name cannot be empty", + details={"name": raw_key}, + ) + + normalized = key.upper() + if not _ENV_KEY_PATTERN.match(normalized): + raise LspError( + "Environment variable name contains invalid characters", + details={"name": key}, + ) + + if normalized in _DISALLOWED_ENV_VARS: + raise LspError( + "Disallowed environment variable for LSP server", + details={"name": key}, + ) + + if raw_value is None: + raise LspError( + f"Environment variable '{key}' must have a value", + details={"name": key}, + ) + + value = str(raw_value) + if "\x00" in value: + raise LspError( + "Environment variable value contains null byte", + details={"name": key}, + ) + + sanitized[key] = value + + return sanitized + def _read_one_message(self, timeout: float) -> dict[str, Any] | None: """Parse a single ``Content-Length`` framed JSON-RPC message.""" assert self._process is not None -- 2.52.0 From a3cd2adf096c6634cac02ebb35b3939709c77d1c Mon Sep 17 00:00:00 2001 From: HAL9000 Date: Thu, 23 Apr 2026 11:04:48 +0000 Subject: [PATCH 2/4] fix(lsp): apply ruff format to lsp_transport_coverage_steps.py --- features/steps/lsp_transport_coverage_steps.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/features/steps/lsp_transport_coverage_steps.py b/features/steps/lsp_transport_coverage_steps.py index 6c635234d..53790b437 100644 --- a/features/steps/lsp_transport_coverage_steps.py +++ b/features/steps/lsp_transport_coverage_steps.py @@ -532,6 +532,6 @@ def step_ltcov_env_contains(context: Context, key: str, value: str) -> None: @then("ltcov Popen should not have been invoked") def step_ltcov_popen_not_called(context: Context) -> None: - assert ( - context.ltcov_popen_call_count == 0 - ), f"Expected no Popen calls, got {context.ltcov_popen_call_count}" + assert context.ltcov_popen_call_count == 0, ( + f"Expected no Popen calls, got {context.ltcov_popen_call_count}" + ) -- 2.52.0 From 88c7699f13a4c911312973b0bf89e510d915998e Mon Sep 17 00:00:00 2001 From: CleverThis Date: Thu, 4 Jun 2026 16:26:27 -0400 Subject: [PATCH 3/4] fix(lsp): allow PYTHONPATH in LSP subprocess environment PYTHONPATH is a legitimate configuration variable required by LSP servers such as Pyright and Pylance to locate project packages. Removing it from _DISALLOWED_ENV_VARS restores this capability as specified in the issue security advisory, which explicitly listed PYTHONPATH as an allowed variable. ISSUES CLOSED: #7184 --- src/cleveragents/lsp/transport.py | 1 - 1 file changed, 1 deletion(-) diff --git a/src/cleveragents/lsp/transport.py b/src/cleveragents/lsp/transport.py index 9f21bc440..3a59e3940 100644 --- a/src/cleveragents/lsp/transport.py +++ b/src/cleveragents/lsp/transport.py @@ -55,7 +55,6 @@ _DISALLOWED_ENV_VARS = { "PATH", "PATHEXT", "PYTHONHOME", - "PYTHONPATH", "PYTHONSTARTUP", "PYTHONBREAKPOINT", "PYTHONUSERBASE", -- 2.52.0 From 014673bcf92790196dc94f9cf8371fc4a20af1cd Mon Sep 17 00:00:00 2001 From: CleverThis Date: Thu, 4 Jun 2026 19:46:40 -0400 Subject: [PATCH 4/4] fix(tests): capture real subprocess.Popen for spec at import time MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The ltcov test step `step_ltcov_popen_capture_env` patches `cleveragents.lsp.transport.subprocess.Popen` and invokes `_make_mock_process()` from inside the `side_effect` callback. Because Python module objects are singletons, that patch also replaces `subprocess.Popen` globally. When `make_mock_process` then evaluated `MagicMock(spec=subprocess.Popen)`, the spec target was itself a MagicMock, and `mock` raised "Cannot spec a Mock object" — failing the `ltcov start merges sanitized environment variables` scenario. Resolve the real `Popen` class once at module import time and use that captured reference as the spec, so `make_mock_process` works regardless of whether a patch is active at call time. --- features/steps/_ltcov_helpers.py | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) diff --git a/features/steps/_ltcov_helpers.py b/features/steps/_ltcov_helpers.py index 8554fcd69..2c150313d 100644 --- a/features/steps/_ltcov_helpers.py +++ b/features/steps/_ltcov_helpers.py @@ -11,6 +11,17 @@ import subprocess from typing import Any from unittest.mock import MagicMock +# Capture the real Popen class at import time. Tests routinely patch +# ``cleveragents.lsp.transport.subprocess.Popen``; because Python module +# objects are singletons, that patch also replaces ``subprocess.Popen`` +# globally for the duration of the test. If ``make_mock_process`` is +# invoked from inside a ``side_effect`` callback while the patch is +# active, ``subprocess.Popen`` is itself a MagicMock — and +# ``MagicMock(spec=)`` raises "Cannot spec a Mock object". +# Resolve the real class once here so the spec is stable regardless of +# whether a patch is active at call time. +_REAL_POPEN = subprocess.Popen + def make_mock_process( *, @@ -35,7 +46,7 @@ def make_mock_process( Returns: A fully configured ``MagicMock`` with ``subprocess.Popen`` spec. """ - proc = MagicMock(spec=subprocess.Popen) + proc = MagicMock(spec=_REAL_POPEN) proc.pid = pid proc.poll.return_value = poll_return proc.returncode = returncode -- 2.52.0