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