diff --git a/tests/tools/test_terminal_none_command_guard.py b/tests/tools/test_terminal_none_command_guard.py new file mode 100644 index 0000000000..05455836d1 --- /dev/null +++ b/tests/tools/test_terminal_none_command_guard.py @@ -0,0 +1,21 @@ +"""Regression tests for invalid/None terminal command handling.""" + +import json + +from tools.terminal_tool import _transform_sudo_command, terminal_tool + + +def test_transform_sudo_command_none_returns_cleanly(): + transformed, sudo_stdin = _transform_sudo_command(None) + + assert transformed is None + assert sudo_stdin is None + + +def test_terminal_tool_none_command_returns_clean_error(): + result = json.loads(terminal_tool(None)) # type: ignore[arg-type] + + assert result["exit_code"] == -1 + assert result["status"] == "error" + assert "expected string" in result["error"].lower() + assert "nonetype" in result["error"].lower() diff --git a/tools/terminal_tool.py b/tools/terminal_tool.py index 183e898335..96a1147759 100644 --- a/tools/terminal_tool.py +++ b/tools/terminal_tool.py @@ -327,7 +327,19 @@ def _prompt_for_sudo_password(timeout_seconds: int = 45) -> str: del os.environ["HERMES_SPINNER_PAUSE"] -def _transform_sudo_command(command: str) -> tuple[str, str | None]: +def _safe_command_preview(command: Any, limit: int = 200) -> str: + """Return a log-safe preview for possibly-invalid command values.""" + if command is None: + return "" + if isinstance(command, str): + return command[:limit] + try: + return repr(command)[:limit] + except Exception: + return f"<{type(command).__name__}>" + + +def _transform_sudo_command(command: str | None) -> tuple[str | None, str | None]: """ Transform sudo commands to use -S flag if SUDO_PASSWORD is available. @@ -365,6 +377,9 @@ def _transform_sudo_command(command: str) -> tuple[str, str | None]: import re # Check if command even contains sudo + if command is None: + return None, None + if not re.search(r'\bsudo\b', command): return command, None # No sudo in command, nothing to do @@ -1050,6 +1065,18 @@ def terminal_tool( # Note: force parameter is internal only, not exposed to model API """ try: + if not isinstance(command, str): + logger.warning( + "Rejected invalid terminal command value: %s", + type(command).__name__, + ) + return json.dumps({ + "output": "", + "exit_code": -1, + "error": f"Invalid command: expected string, got {type(command).__name__}", + "status": "error", + }, ensure_ascii=False) + # Get configuration config = _get_env_config() env_type = config["env_type"] @@ -1207,7 +1234,7 @@ def terminal_tool( workdir_error = _validate_workdir(workdir) if workdir_error: logger.warning("Blocked dangerous workdir: %s (command: %s)", - workdir[:200], command[:200]) + workdir[:200], _safe_command_preview(command)) return json.dumps({ "output": "", "exit_code": -1, @@ -1347,12 +1374,12 @@ def terminal_tool( retry_count += 1 wait_time = 2 ** retry_count logger.warning("Execution error, retrying in %ds (attempt %d/%d) - Command: %s - Error: %s: %s - Task: %s, Backend: %s", - wait_time, retry_count, max_retries, command[:200], type(e).__name__, e, effective_task_id, env_type) + wait_time, retry_count, max_retries, _safe_command_preview(command), type(e).__name__, e, effective_task_id, env_type) time.sleep(wait_time) continue logger.error("Execution failed after %d retries - Command: %s - Error: %s: %s - Task: %s, Backend: %s", - max_retries, command[:200], type(e).__name__, e, effective_task_id, env_type) + max_retries, _safe_command_preview(command), type(e).__name__, e, effective_task_id, env_type) return json.dumps({ "output": "", "exit_code": -1,