security(tools): eliminate RCE in template loader eval() and shell=True (#216)

`python`-action tool templates evaluated expressions with `eval()` under a
restricted `__builtins__`. That sandbox is escapable via attribute walks like
`str.__class__.__mro__[-1].__subclasses__()`, reaching `object.__subclasses__()`
and arbitrary code. `shell`-action templates interpolated parameters into a
string and ran it with `shell=True`, so a value like `; rm -rf ~` or
`$(curl evil)` executed in the host shell.

Fixes:
- Replace `eval()` with a small AST interpreter (`safe_eval_expr`) that
  implements an explicit node allowlist — literals, names, arithmetic/boolean/
  comparison ops, ternaries, subscripts, container literals, and calls to a
  fixed set of builtins only. Attribute access, lambdas, comprehensions, and
  dunder names have no implementation and raise `ValueError`, so the escape
  vectors are unreachable by construction. No `eval`/`exec` remains.
- Shell action: tokenize the FIXED template with `shlex.split` first, then
  substitute params into individual argv elements and run with `shell=False`.
  Injected metacharacters become inert literal arguments.

All shipped builtin templates (`str(float(value))`, `str(input) if input
else ...`, etc.) continue to work. Verified empirically: every known escape
payload is rejected and a `; touch <marker>` / `$(touch <marker>)` /
backtick injection never creates the marker file.

Closes #216

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

src/openjarvis/tools/templates/loader.py

@@ -2,11 +2,14 @@
 
 from __future__ import annotations
 
+import ast
 import json
 import logging
+import operator
+import shlex
 import subprocess
 from pathlib import Path
-from typing import Any, Dict, List, Optional
+from typing import Any, Callable, Dict, List, Optional
 
 from openjarvis.core.types import ToolResult
 from openjarvis.tools._stubs import BaseTool, ToolSpec
@@ -19,6 +22,150 @@ except ModuleNotFoundError:
     import tomli as tomllib  # type: ignore[no-redef]
 
 
+# --- Safe expression evaluation for ``python``-action templates -----------
+# Templates may embed a Python expression (e.g. ``"str(float(value))"``).
+# A bare ``eval()`` — even with a restricted ``__builtins__`` — is trivially
+# escapable via attribute walks such as
+# ``str.__class__.__mro__[-1].__subclasses__()``, which reach
+# ``object.__subclasses__()`` and from there arbitrary code execution.
+#
+# Instead of calling ``eval()`` at all, we parse the expression and walk the
+# AST with a small interpreter that implements only an explicit allowlist of
+# node types. Attribute access, lambdas, comprehensions, dunder names, and
+# calls to anything other than the whitelisted builtins simply have no
+# implementation and raise ``ValueError`` — the escape vectors are
+# unreachable by construction.
+
+_SAFE_EVAL_FUNCS: Dict[str, Callable[..., Any]] = {
+    "str": str,
+    "int": int,
+    "float": float,
+    "bool": bool,
+    "len": len,
+    "abs": abs,
+    "min": min,
+    "max": max,
+    "round": round,
+    "sorted": sorted,
+    "list": list,
+    "dict": dict,
+    "tuple": tuple,
+    "set": set,
+}
+
+_BIN_OPS: Dict[type, Callable[[Any, Any], Any]] = {
+    ast.Add: operator.add,
+    ast.Sub: operator.sub,
+    ast.Mult: operator.mul,
+    ast.Div: operator.truediv,
+    ast.FloorDiv: operator.floordiv,
+    ast.Mod: operator.mod,
+    ast.Pow: operator.pow,
+}
+
+_UNARY_OPS: Dict[type, Callable[[Any], Any]] = {
+    ast.UAdd: operator.pos,
+    ast.USub: operator.neg,
+    ast.Not: operator.not_,
+    ast.Invert: operator.invert,
+}
+
+_CMP_OPS: Dict[type, Callable[[Any, Any], bool]] = {
+    ast.Eq: operator.eq,
+    ast.NotEq: operator.ne,
+    ast.Lt: operator.lt,
+    ast.LtE: operator.le,
+    ast.Gt: operator.gt,
+    ast.GtE: operator.ge,
+    ast.Is: operator.is_,
+    ast.IsNot: operator.is_not,
+    ast.In: lambda a, b: a in b,
+    ast.NotIn: lambda a, b: a not in b,
+}
+
+
+def _eval_node(node: ast.AST, names: Dict[str, Any]) -> Any:
+    """Recursively evaluate a single whitelisted AST node."""
+    if isinstance(node, ast.Expression):
+        return _eval_node(node.body, names)
+    if isinstance(node, ast.Constant):
+        return node.value
+    if isinstance(node, ast.Name):
+        if node.id.startswith("__"):
+            raise ValueError(f"disallowed name: {node.id}")
+        if node.id in names:
+            return names[node.id]
+        if node.id in _SAFE_EVAL_FUNCS:
+            return _SAFE_EVAL_FUNCS[node.id]
+        raise ValueError(f"unknown name: {node.id}")
+    if isinstance(node, ast.BinOp) and type(node.op) in _BIN_OPS:
+        return _BIN_OPS[type(node.op)](
+            _eval_node(node.left, names), _eval_node(node.right, names)
+        )
+    if isinstance(node, ast.UnaryOp) and type(node.op) in _UNARY_OPS:
+        return _UNARY_OPS[type(node.op)](_eval_node(node.operand, names))
+    if isinstance(node, ast.BoolOp):
+        result: Any = isinstance(node.op, ast.And)
+        for value in node.values:
+            result = _eval_node(value, names)
+            if isinstance(node.op, ast.And) and not result:
+                return result
+            if isinstance(node.op, ast.Or) and result:
+                return result
+        return result
+    if isinstance(node, ast.Compare):
+        left = _eval_node(node.left, names)
+        for op, comparator in zip(node.ops, node.comparators):
+            if type(op) not in _CMP_OPS:
+                raise ValueError(f"disallowed comparison: {type(op).__name__}")
+            right = _eval_node(comparator, names)
+            if not _CMP_OPS[type(op)](left, right):
+                return False
+            left = right
+        return True
+    if isinstance(node, ast.IfExp):
+        branch = node.body if _eval_node(node.test, names) else node.orelse
+        return _eval_node(branch, names)
+    if isinstance(node, ast.Call):
+        if node.keywords:
+            raise ValueError("keyword arguments are not allowed")
+        if not isinstance(node.func, ast.Name) or node.func.id not in _SAFE_EVAL_FUNCS:
+            raise ValueError("only calls to whitelisted builtins are allowed")
+        func = _SAFE_EVAL_FUNCS[node.func.id]
+        return func(*[_eval_node(a, names) for a in node.args])
+    if isinstance(node, ast.Subscript):
+        return _eval_node(node.value, names)[_eval_node(node.slice, names)]
+    if isinstance(node, ast.Slice):
+        return slice(
+            _eval_node(node.lower, names) if node.lower else None,
+            _eval_node(node.upper, names) if node.upper else None,
+            _eval_node(node.step, names) if node.step else None,
+        )
+    if isinstance(node, ast.List):
+        return [_eval_node(e, names) for e in node.elts]
+    if isinstance(node, ast.Tuple):
+        return tuple(_eval_node(e, names) for e in node.elts)
+    if isinstance(node, ast.Set):
+        return {_eval_node(e, names) for e in node.elts}
+    if isinstance(node, ast.Dict):
+        return {
+            _eval_node(k, names): _eval_node(v, names)
+            for k, v in zip(node.keys, node.values)
+        }
+    raise ValueError(f"disallowed expression element: {type(node).__name__}")
+
+
+def safe_eval_expr(expr: str, names: Dict[str, Any]) -> Any:
+    """Safely evaluate a template ``python``-action expression.
+
+    Parses *expr* and interprets it against the allowlist in
+    :func:`_eval_node`. No ``eval``/``exec`` is used, and the only callables
+    reachable are :data:`_SAFE_EVAL_FUNCS` plus the supplied *names* (the tool
+    parameters). Raises ``ValueError`` / ``SyntaxError`` on anything unsafe.
+    """
+    return _eval_node(ast.parse(expr, mode="eval"), names)
+
+
 class ToolTemplate(BaseTool):
     """A tool dynamically constructed from a TOML template definition."""
 
@@ -74,22 +221,14 @@ class ToolTemplate(BaseTool):
                 content="No expression defined.",
                 success=False,
             )
-        # Safe evaluation with params available
-        safe_builtins = {
-            "str": str,
-            "int": int,
-            "float": float,
-            "len": len,
-            "sorted": sorted,
-            "list": list,
-            "dict": dict,
-            "json": json,
-        }
-        result = eval(  # noqa: S307
-            expr,
-            {"__builtins__": safe_builtins},
-            params,
-        )
+        try:
+            result = safe_eval_expr(expr, params)
+        except (ValueError, SyntaxError) as exc:
+            return ToolResult(
+                tool_name=self._name,
+                content=f"Rejected unsafe or invalid expression: {exc}",
+                success=False,
+            )
         return ToolResult(
             tool_name=self._name,
             content=str(result),
@@ -98,19 +237,37 @@ class ToolTemplate(BaseTool):
 
     def _execute_shell(self, params: Dict[str, Any]) -> ToolResult:
         """Execute a shell command (requires code:execute capability)."""
-        cmd = self._action.get("command", "")
-        if not cmd:
+        cmd_template = self._action.get("command", "")
+        if not cmd_template:
             return ToolResult(
                 tool_name=self._name,
                 content="No command defined.",
                 success=False,
             )
-        # Substitute params into command
-        for key, val in params.items():
-            cmd = cmd.replace(f"{{{key}}}", str(val))
-        result = subprocess.run(  # noqa: S602, S603
-            cmd,
-            shell=True,
+        # Tokenize the FIXED template command first, then substitute params
+        # into individual argv elements and run WITHOUT a shell. A parameter
+        # value such as ``; rm -rf ~`` or ``$(curl evil)`` becomes a single
+        # literal argument rather than shell syntax, so it cannot inject.
+        try:
+            tokens = shlex.split(cmd_template)
+        except ValueError as exc:
+            return ToolResult(
+                tool_name=self._name,
+                content=f"Invalid command template: {exc}",
+                success=False,
+            )
+        argv = [
+            self._substitute(token, params) for token in tokens
+        ]
+        if not argv:
+            return ToolResult(
+                tool_name=self._name,
+                content="Empty command.",
+                success=False,
+            )
+        result = subprocess.run(  # noqa: S603
+            argv,
+            shell=False,
             capture_output=True,
             text=True,
             timeout=30,
@@ -122,6 +279,13 @@ class ToolTemplate(BaseTool):
             success=result.returncode == 0,
         )
 
+    @staticmethod
+    def _substitute(token: str, params: Dict[str, Any]) -> str:
+        """Replace ``{key}`` placeholders in a single argv token."""
+        for key, val in params.items():
+            token = token.replace(f"{{{key}}}", str(val))
+        return token
+
     def _execute_transform(self, params: Dict[str, Any]) -> ToolResult:
         """Execute a data transformation."""
         transform = self._action.get("transform", "identity")

tests/tools/test_template_loader_security.py

@@ -0,0 +1,126 @@
+"""Security regression tests for the tool-template loader (issue #216).
+
+`python`-action templates must not be able to escape the expression sandbox
+(no attribute walks reaching `object.__subclasses__()`), and `shell`-action
+templates must not allow command injection through parameter substitution.
+"""
+
+from __future__ import annotations
+
+import os
+
+import pytest
+
+from openjarvis.tools.templates.loader import (
+    ToolTemplate,
+    discover_templates,
+    safe_eval_expr,
+)
+
+# Expressions that previously reached arbitrary code execution via the
+# sandbox escape, plus other dangerous constructs. All must raise.
+ESCAPE_EXPRESSIONS = [
+    "str.__class__.__mro__[-1].__subclasses__()",
+    "().__class__.__bases__[0].__subclasses__()",
+    "__import__('os').system('id')",
+    "[c for c in ().__class__.__base__.__subclasses__()]",
+    "(lambda: 1)()",
+    "open('/etc/passwd').read()",
+    "globals()",
+    "().__class__",
+]
+
+
+@pytest.mark.parametrize("expr", ESCAPE_EXPRESSIONS)
+def test_escape_expressions_are_rejected(expr: str) -> None:
+    with pytest.raises((ValueError, SyntaxError)):
+        safe_eval_expr(expr, {})
+
+
+@pytest.mark.parametrize(
+    ("expr", "names", "expected"),
+    [
+        ("str(float(value))", {"value": 3}, "3.0"),
+        ("str(input)", {"input": "hi"}, "hi"),
+        ("str(input) if input else 'no input'", {"input": ""}, "no input"),
+        ("len(input) * 2 + 1", {"input": "abcd"}, 9),
+        ("sorted(input)[0]", {"input": [3, 1, 2]}, 1),
+        ("max(a, b)", {"a": 5, "b": 9}, 9),
+        ("input[1:3]", {"input": "abcdef"}, "bc"),
+    ],
+)
+def test_legitimate_expressions_still_work(expr, names, expected) -> None:
+    assert safe_eval_expr(expr, names) == expected
+
+
+def test_python_action_rejects_escape_via_execute() -> None:
+    tmpl = ToolTemplate(
+        {
+            "name": "evil",
+            "action": {
+                "type": "python",
+                "expression": "str.__class__.__mro__[-1].__subclasses__()",
+            },
+        }
+    )
+    result = tmpl.execute()
+    assert result.success is False
+    assert "unsafe" in result.content.lower() or "disallowed" in result.content.lower()
+
+
+def test_shell_action_neutralizes_injection(tmp_path) -> None:
+    """A `;`-style injection in a parameter must not run as shell syntax."""
+    marker = tmp_path / "pwned"
+    tmpl = ToolTemplate(
+        {
+            "name": "echoer",
+            "action": {"type": "shell", "command": "echo {msg}"},
+        }
+    )
+    result = tmpl.execute(msg=f"hello; touch {marker}")
+    # The injected command text is echoed back literally, not executed.
+    assert "hello" in result.content
+    assert not marker.exists(), "command injection executed — marker was created"
+
+
+def test_shell_action_passes_values_as_single_argument() -> None:
+    """Whitespace in a value stays one argv element (no word-splitting)."""
+    tmpl = ToolTemplate(
+        {
+            "name": "echoer",
+            "action": {"type": "shell", "command": "echo {msg}"},
+        }
+    )
+    result = tmpl.execute(msg="a b c")
+    assert result.content == "a b c"
+
+
+def test_builtin_templates_still_load_and_run() -> None:
+    """Every shipped builtin template loads and its action executes safely."""
+    templates = discover_templates()
+    assert templates, "no builtin templates discovered"
+    for tmpl in templates:
+        # Exercise with a representative input; should not raise.
+        result = tmpl.execute(input="hello", value=1.5, text="hi")
+        assert result.tool_name == tmpl.spec.name
+
+
+def test_shell_action_command_substitution_is_inert(tmp_path) -> None:
+    """Backtick / command-substitution payloads are inert too."""
+    marker = tmp_path / "sub"
+    tmpl = ToolTemplate(
+        {
+            "name": "echoer",
+            "action": {"type": "shell", "command": "echo {msg}"},
+        }
+    )
+    tmpl.execute(msg=f"$(touch {marker})")
+    assert not marker.exists()
+    tmpl.execute(msg=f"`touch {marker}`")
+    assert not marker.exists()
+
+
+def test_attribute_access_unavailable_even_for_param_objects() -> None:
+    """Even if a param holds a module, attribute access stays unreachable."""
+    with pytest.raises((ValueError, SyntaxError)):
+        safe_eval_expr("payload.__class__", {"payload": os})