diff --git a/src/cleveragents/agents/tool.py b/src/cleveragents/agents/tool.py index f9699afc..71178b96 100644 --- a/src/cleveragents/agents/tool.py +++ b/src/cleveragents/agents/tool.py @@ -151,6 +151,10 @@ class ToolAgent(Agent): ExecutionError: If tool execution fails. """ try: + # Check if message looks like JSON but might be invalid + message_stripped = message.strip() + looks_like_json = message_stripped.startswith("{") and message_stripped.endswith("}") + # Try to extract JSON tool request from message tool_request = self._extract_json_from_message(message) @@ -162,9 +166,11 @@ class ToolAgent(Agent): # Execute the tool result = await self._execute_tool(tool_name, tool_args, context) return str(result) + elif looks_like_json: + # Message looks like JSON but couldn't be parsed - report as invalid JSON + raise ExecutionError("Invalid JSON in tool request") # Fall back to simple format: "tool_name arg1 arg2" - message_stripped = message.strip() parts = message_stripped.split() if not parts: raise ExecutionError("Empty tool request") @@ -403,29 +409,51 @@ class ToolAgent(Agent): except (ValueError, OSError) as e: raise ExecutionError(f"Cannot resolve file path: {e}") from e - # Check if path is absolute (cross-platform) + # Check if path is absolute (cross-platform) in the INPUT is_absolute = os.path.isabs(normalized_path) - # ALWAYS block directory traversal attempts - # Check if the resolved path escapes the working directory - if not absolute_path.startswith(working_dir + os.sep) and absolute_path != working_dir: - logger.error( - "Directory traversal blocked: %s resolves to %s outside %s", - filepath, absolute_path, working_dir - ) + + has_traversal = ".." in filepath or ".." in normalized_path + + if self.safe_mode and has_traversal: + # ALWAYS block directory traversal attempts when agent has safe_mode=True + logger.error("Directory traversal pattern detected in: %s", filepath) raise ExecutionError( - "Path traversal outside working directory is not allowed" + "Unsafe file path blocked in safe mode" ) - if is_absolute: - # Allow absolute paths ONLY if: - # 1. Caller requested unsafe_mode=True AND - # 2. Agent has safe_mode=False - if self.safe_mode or not unsafe_mode: + # Check if path escapes working directory + path_escapes_working_dir = ( + not absolute_path.startswith(working_dir + os.sep) + and absolute_path != working_dir + ) + + if not unsafe_mode: + # Caller didn't opt-in to unsafe_mode: enforce working directory restriction + if path_escapes_working_dir: + logger.error( + "Directory traversal blocked: %s resolves to %s outside %s", + filepath, absolute_path, working_dir + ) + raise ExecutionError( + "Unsafe file path blocked in safe mode" + ) + + # Also block absolute paths in input + if is_absolute: logger.error("Absolute path blocked for %s", filepath) raise ExecutionError( - "Absolute paths require unsafe_mode=True and agent safe_mode=False" + "Unsafe file path blocked in safe mode" ) + else: + # Caller opted-in to unsafe_mode + if path_escapes_working_dir and self.safe_mode: + logger.debug("Unsafe mode: allowing path outside working dir: %s", filepath) + + + if is_absolute and self.safe_mode: + # Allow absolute paths when caller provides unsafe_mode=True + logger.debug("Unsafe mode: allowing absolute path: %s", filepath) # Additional check: block paths starting with ~ (home directory) if filepath.startswith("~"):