fix: conservative trust-boundary hardening (#540)

call_method dunder denylist, instantiate non-estimator rejection, run_command
doc fix + output cap. Addresses #540 (BUG-09 sandboxing left for maintainers).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: Shashankss1205

src/sktime_mcp/runtime/executor.py

@@ -87,6 +87,29 @@ def _to_period_index_if_possible(obj: Any) -> Any:
 # flood the client with ~30KB+ of inline JSON.
 _MAX_PREDICTION_ROWS = 500
 
+# Dunder methods that are safe and useful to call via call_method (e.g. __call__
+# for callable metrics/aligners). Everything else starting with "_" is blocked
+# (BUG-11) — notably __reduce__/__class__/__getattribute__ and private methods.
+_ALLOWED_DUNDERS = frozenset({"__call__", "__len__", "__repr__", "__str__"})
+
+
+def _is_sktime_object(obj: Any) -> bool:
+    """True if *obj* is a genuine sktime estimator/object, not a bare value.
+
+    craft evaluates arbitrary specs, so a spec like "42" returns an int. Such
+    non-objects should not receive an estimator handle (BUG-10). We accept
+    anything deriving from skbase's BaseObject, falling back to a duck-typed
+    check for get_params + a scitype tag.
+    """
+    try:
+        from skbase.base import BaseObject
+
+        if isinstance(obj, BaseObject):
+            return True
+    except Exception:  # pragma: no cover - skbase always present with sktime
+        pass
+    return hasattr(obj, "get_params") and hasattr(obj, "get_class_tag")
+
 
 def _cap_prediction_rows(result: dict) -> tuple[dict, dict | None]:
     """Cap an index-keyed prediction dict, returning (capped, truncation_note)."""
@@ -326,6 +349,19 @@ def mock_all_estimators(*args, **kwargs):
             finally:
                 _craft_module.all_estimators = original_all
 
+            # Reject specs that don't produce an sktime object — e.g. "42",
+            # "[1,2,3]", "None" otherwise got est_ handles that failed
+            # confusingly downstream (BUG-10).
+            if not _is_sktime_object(instance):
+                return {
+                    "success": False,
+                    "error": (
+                        f"Spec did not produce an sktime estimator, got "
+                        f"{type(instance).__name__}. Provide a craft spec such as "
+                        "'NaiveForecaster(sp=12)' or 'Detrender() * ARIMA()'."
+                    ),
+                }
+
             estimator_name = type(instance).__name__
             handle_id = self._handle_manager.create_handle(
                 estimator_name=estimator_name,
@@ -732,6 +768,18 @@ def call_method(
         except KeyError:
             return {"success": False, "error": self._handle_manager.describe_missing(handle_id)}
 
+        # Block private/dunder methods: they are not part of the estimator API
+        # and expose internals — e.g. __reduce__ dumps __dict__ including the
+        # fitted _y/_X training data to any caller (BUG-11).
+        if method_name.startswith("_") and method_name not in _ALLOWED_DUNDERS:
+            return {
+                "success": False,
+                "error": (
+                    f"Method '{method_name}' is private and not callable via call_method. "
+                    "Only public estimator methods are exposed."
+                ),
+            }
+
         if not hasattr(instance, method_name):
             obj_type = getattr(instance, "get_class_tag", lambda x, y: "")("object_type", "")
             return {

src/sktime_mcp/server.py

@@ -958,16 +958,18 @@ async def list_tools() -> list[Tool]:
         Tool(
             name="run_command",
             description=(
-                "Run an arbitrary CLI/bash command inside the sktime container. "
-                "Use this to install missing python packages (e.g., 'pip install mlflow') "
-                "or inspect the file system."
+                "Run an arbitrary CLI/bash command on the HOST machine, in the server's "
+                "working directory, as the user that launched the server (not a container "
+                "or sandbox). Use it to install packages into the server's environment "
+                "(e.g. the server's python -m pip install mlflow) or inspect the filesystem. "
+                "Output is capped; a 'truncated' flag indicates when it was shortened."
             ),
             inputSchema={
                 "type": "object",
                 "properties": {
                     "command": {
                         "type": "string",
-                        "description": "The bash command to run",
+                        "description": "The shell command to run (executed via /bin/sh -c).",
                     },
                 },
                 "required": ["command"],

src/sktime_mcp/tools/run_command.py

@@ -1,22 +1,51 @@
 import subprocess
 from typing import Any
 
+# Cap on returned output so a command like `seq 1 100000` (688k chars observed)
+# can't overflow the client (BUG-23).
+_MAX_OUTPUT_CHARS = 20_000
+
+
+def _truncate(text: str) -> tuple[str, bool]:
+    if len(text) <= _MAX_OUTPUT_CHARS:
+        return text, False
+    head = _MAX_OUTPUT_CHARS // 2
+    tail = _MAX_OUTPUT_CHARS - head
+    omitted = len(text) - _MAX_OUTPUT_CHARS
+    return (
+        f"{text[:head]}\n...[{omitted} characters truncated]...\n{text[-tail:]}",
+        True,
+    )
+
 
 def run_command_tool(command: str) -> dict[str, Any]:
     """
-    Run an arbitrary CLI/bash command inside the sktime-mcp container.
+    Run an arbitrary CLI/bash command on the host, in the server's working
+    directory, as the user that launched the server.
+
+    Note: this executes on the host machine (not a container or sandbox) with
+    the server's own permissions. Output is capped; see ``truncated``.
     """
     try:
         result = subprocess.run(command, shell=True, capture_output=True, text=True, timeout=120)
         output = result.stdout
         if result.stderr:
             output += f"\nErrors:\n{result.stderr}"
 
-        return {
+        output, truncated = _truncate(output.strip())
+        response = {
             "success": result.returncode == 0,
-            "output": output.strip(),
+            "output": output,
             "returncode": result.returncode,
+            "truncated": truncated,
         }
+        # Every other tool reports failures under "error"; do the same on a
+        # non-zero exit so callers can branch on it uniformly.
+        if result.returncode != 0:
+            response["error"] = (
+                result.stderr.strip() or f"Command exited with code {result.returncode}"
+            )
+        return response
     except subprocess.TimeoutExpired:
         return {"success": False, "error": "Command execution timed out after 120 seconds."}
     except Exception as e:

tests/test_evaluate_scitype.py

@@ -32,12 +32,12 @@ def test_rejects_transformer():
 
 
 def test_rejects_non_estimator():
-    res = instantiate_tool(spec="42")
-    handle = res["handle"]
+    # instantiate now blocks non-estimators (BUG-10), so inject an int handle
+    # directly to exercise evaluate's own object_type guard.
+    handle = get_handle_manager().create_handle("int", 42, {})
     try:
         out = evaluate_tool(estimator_handle=handle, y="airline", cv_folds=3)
         assert not out["success"]
-        # int handle has no forecaster object_type -> rejected (here or by scitype)
         assert "forecaster" in out["error"].lower() or "series" in out["error"].lower()
     finally:
         _release(handle)

tests/test_security_hardening.py

@@ -0,0 +1,87 @@
+"""Conservative trust-boundary hardening (#540).
+
+Covers the non-controversial parts: block private/dunder methods in
+call_method (BUG-11), reject non-estimator instantiate results (BUG-10), and
+cap run_command output (BUG-23). instantiate sandboxing (BUG-09) is
+intentionally out of scope — the server already exposes run_command.
+"""
+
+import contextlib
+
+import pytest
+
+from sktime_mcp.runtime.executor import get_executor
+from sktime_mcp.runtime.handles import get_handle_manager
+from sktime_mcp.tools.instantiate import instantiate_tool
+from sktime_mcp.tools.run_command import run_command_tool
+
+
+def _release(handle):
+    with contextlib.suppress(KeyError):
+        get_handle_manager().release_handle(handle)
+
+
+class TestCallMethodDenylist:
+    def test_reduce_blocked(self):
+        h = instantiate_tool(spec="NaiveForecaster()")["handle"]
+        try:
+            res = get_executor().call_method(handle_id=h, method_name="__reduce__", kwargs={})
+            assert not res["success"]
+            assert "private" in res["error"].lower()
+        finally:
+            _release(h)
+
+    @pytest.mark.parametrize("m", ["__class__", "__getattribute__", "_get_class_flags", "__init__"])
+    def test_private_methods_blocked(self, m):
+        h = instantiate_tool(spec="NaiveForecaster()")["handle"]
+        try:
+            res = get_executor().call_method(handle_id=h, method_name=m, kwargs={})
+            assert not res["success"]
+            assert "private" in res["error"].lower()
+        finally:
+            _release(h)
+
+    def test_public_method_still_works(self):
+        h = instantiate_tool(spec="NaiveForecaster(sp=12)")["handle"]
+        try:
+            res = get_executor().call_method(handle_id=h, method_name="get_params", kwargs={})
+            assert res["success"]
+            assert res["result"]["sp"] == 12
+        finally:
+            _release(h)
+
+
+class TestInstantiateTypeCheck:
+    @pytest.mark.parametrize("spec", ["42", "[1, 2, 3]", "'just a string'", "None"])
+    def test_non_estimator_rejected(self, spec):
+        res = instantiate_tool(spec=spec)
+        assert not res["success"]
+        assert "did not produce an sktime estimator" in res["error"]
+
+    def test_real_estimator_still_instantiates(self):
+        res = instantiate_tool(spec="NaiveForecaster(sp=12)")
+        try:
+            assert res["success"]
+        finally:
+            _release(res.get("handle"))
+
+
+class TestRunCommandCap:
+    def test_large_output_truncated(self):
+        res = run_command_tool("seq 1 100000")
+        assert res["success"]
+        assert res["truncated"] is True
+        assert len(res["output"]) < 25_000
+        assert "truncated" in res["output"]
+
+    def test_small_output_not_truncated(self):
+        res = run_command_tool("echo hello")
+        assert res["success"]
+        assert res["truncated"] is False
+        assert res["output"] == "hello"
+
+    def test_nonzero_exit_has_error_key(self):
+        res = run_command_tool("exit 3")
+        assert not res["success"]
+        assert res["returncode"] == 3
+        assert "error" in res