diff --git a/src/openjarvis/tools/templates/loader.py b/src/openjarvis/tools/templates/loader.py index 014c736a..8b2013a9 100644 --- a/src/openjarvis/tools/templates/loader.py +++ b/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") diff --git a/tests/tools/test_template_loader_security.py b/tests/tools/test_template_loader_security.py new file mode 100644 index 00000000..b2c0e7e8 --- /dev/null +++ b/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})