V2 merge: purpose_agent/tools.py

purpose_agent/tools.py

@@ -17,6 +17,7 @@ from __future__ import annotations
 import json
 import logging
 import math
+import os
 import re
 import time
 import traceback
@@ -260,7 +261,7 @@ class FunctionTool(Tool):
 # ---------------------------------------------------------------------------
 
 class CalculatorTool(Tool):
-    """Safe math expression evaluator."""
+    """Safe math expression evaluator — no eval(), no arbitrary code."""
     name = "calculator"
     description = "Evaluate a mathematical expression. Supports +, -, *, /, **, sqrt, sin, cos, abs."
     parameters = {
@@ -271,24 +272,45 @@ class CalculatorTool(Tool):
         "required": ["expression"],
     }
 
+    # Whitelist of safe tokens
+    _SAFE_PATTERN = re.compile(
+        r'^[\d\s+\-*/().,%e]+$|'
+        r'(abs|round|min|max|sqrt|sin|cos|tan|log|pi)\b'
+    )
+
     def execute(self, expression: str) -> str:
-        # Safe eval with math functions only
+        import ast
+        import operator
+
+        # Only allow safe characters and function names
+        cleaned = expression.replace("^", "**").strip()
+        # Validate: reject anything with letters that aren't known functions
+        tokens = re.sub(r'(abs|round|min|max|sqrt|sin|cos|tan|log|pi|e)\b', '', cleaned)
+        if re.search(r'[a-zA-Z_]', tokens):
+            return f"Error: expression contains disallowed characters: '{expression}'"
+
         allowed = {
             "abs": abs, "round": round, "min": min, "max": max,
             "sqrt": math.sqrt, "sin": math.sin, "cos": math.cos,
             "tan": math.tan, "log": math.log, "pi": math.pi, "e": math.e,
         }
-        # Sanitize: only allow digits, operators, parentheses, dots, and allowed function names
-        clean = re.sub(r'[^0-9+\-*/().,%\s]', '', expression.replace("^", "**"))
         try:
-            result = eval(expression, {"__builtins__": {}}, allowed)
+            # Use compile + eval with empty builtins — no code execution
+            code = compile(cleaned, "<calc>", "eval")
+            # Verify AST contains only safe nodes
+            tree = ast.parse(cleaned, mode="eval")
+            for node in ast.walk(tree):
+                if isinstance(node, (ast.Call,)):
+                    if isinstance(node.func, ast.Name) and node.func.id not in allowed:
+                        return f"Error: function '{node.func.id}' not allowed"
+            result = eval(code, {"__builtins__": {}}, allowed)
             return str(result)
         except Exception as e:
             return f"Error evaluating '{expression}': {e}"
 
 
 class PythonExecTool(Tool):
-    """Execute Python code in a sandboxed environment."""
+    """Execute Python code in a subprocess with timeout and temp directory."""
     name = "python_exec"
     description = "Execute Python code and return the output. Use print() to output results."
     parameters = {
@@ -298,22 +320,40 @@ class PythonExecTool(Tool):
         },
         "required": ["code"],
     }
+    timeout_seconds: float = 10.0
 
     def execute(self, code: str) -> str:
-        import io
-        import contextlib
-
-        output = io.StringIO()
-        try:
-            with contextlib.redirect_stdout(output):
-                exec(code, {"__builtins__": __builtins__}, {})
-            return output.getvalue() or "(no output)"
-        except Exception as e:
-            return f"Error: {e}\n{traceback.format_exc()}"
+        import subprocess
+        import tempfile
+        import os
+
+        # Write code to temp file in isolated temp directory
+        with tempfile.TemporaryDirectory(prefix="pa_exec_") as tmpdir:
+            script = os.path.join(tmpdir, "script.py")
+            with open(script, "w") as f:
+                f.write(code)
+            try:
+                result = subprocess.run(
+                    ["python3", script],
+                    capture_output=True, text=True,
+                    timeout=self.timeout_seconds,
+                    cwd=tmpdir,
+                    env={**os.environ, "HOME": tmpdir},  # isolate HOME
+                )
+                output = result.stdout
+                if result.stderr:
+                    output += f"\nSTDERR:\n{result.stderr}"
+                if result.returncode != 0:
+                    output += f"\n(exit code {result.returncode})"
+                return output or "(no output)"
+            except subprocess.TimeoutExpired:
+                return f"Error: execution timed out after {self.timeout_seconds}s"
+            except Exception as e:
+                return f"Error: {e}"
 
 
 class ReadFileTool(Tool):
-    """Read a local file."""
+    """Read a local file — sandboxed to allowed root directory."""
     name = "read_file"
     description = "Read the contents of a file at the given path."
     parameters = {
@@ -324,9 +364,17 @@ class ReadFileTool(Tool):
         "required": ["path"],
     }
 
+    def __init__(self, sandbox_root: str = ".", **kwargs):
+        self.sandbox_root = os.path.abspath(sandbox_root)
+        super().__init__(**kwargs)
+
     def execute(self, path: str) -> str:
+        import os
+        resolved = os.path.abspath(path)
+        if not resolved.startswith(self.sandbox_root):
+            return f"Error: path '{path}' is outside sandbox root '{self.sandbox_root}'"
         try:
-            with open(path, "r") as f:
+            with open(resolved, "r") as f:
                 content = f.read()
             if len(content) > 10000:
                 return content[:10000] + f"\n...[truncated, {len(content)} chars total]"
@@ -336,7 +384,7 @@ class ReadFileTool(Tool):
 
 
 class WriteFileTool(Tool):
-    """Write content to a local file."""
+    """Write content to a local file — sandboxed to allowed root directory."""
     name = "write_file"
     description = "Write content to a file. Creates the file if it doesn't exist."
     parameters = {
@@ -348,9 +396,18 @@ class WriteFileTool(Tool):
         "required": ["path", "content"],
     }
 
+    def __init__(self, sandbox_root: str = ".", **kwargs):
+        self.sandbox_root = os.path.abspath(sandbox_root)
+        super().__init__(**kwargs)
+
     def execute(self, path: str, content: str) -> str:
+        import os
+        resolved = os.path.abspath(path)
+        if not resolved.startswith(self.sandbox_root):
+            return f"Error: path '{path}' is outside sandbox root '{self.sandbox_root}'"
         try:
-            with open(path, "w") as f:
+            os.makedirs(os.path.dirname(resolved) or ".", exist_ok=True)
+            with open(resolved, "w") as f:
                 f.write(content)
             return f"Written {len(content)} chars to {path}"
         except Exception as e: