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

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=<MagicMock>)`` 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

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}"
+    )

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,27 @@ _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",
+    "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 +127,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 +149,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 +161,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 +293,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