mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-05-01 00:11:39 +08:00
Compare commits
1 Commits
fix/plugin
...
hermes/her
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
338fe76e63 |
21
tests/tools/test_terminal_none_command_guard.py
Normal file
21
tests/tools/test_terminal_none_command_guard.py
Normal file
@@ -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()
|
||||||
@@ -327,7 +327,19 @@ def _prompt_for_sudo_password(timeout_seconds: int = 45) -> str:
|
|||||||
del os.environ["HERMES_SPINNER_PAUSE"]
|
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 "<None>"
|
||||||
|
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.
|
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
|
import re
|
||||||
|
|
||||||
# Check if command even contains sudo
|
# Check if command even contains sudo
|
||||||
|
if command is None:
|
||||||
|
return None, None
|
||||||
|
|
||||||
if not re.search(r'\bsudo\b', command):
|
if not re.search(r'\bsudo\b', command):
|
||||||
return command, None # No sudo in command, nothing to do
|
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
|
# Note: force parameter is internal only, not exposed to model API
|
||||||
"""
|
"""
|
||||||
try:
|
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
|
# Get configuration
|
||||||
config = _get_env_config()
|
config = _get_env_config()
|
||||||
env_type = config["env_type"]
|
env_type = config["env_type"]
|
||||||
@@ -1207,7 +1234,7 @@ def terminal_tool(
|
|||||||
workdir_error = _validate_workdir(workdir)
|
workdir_error = _validate_workdir(workdir)
|
||||||
if workdir_error:
|
if workdir_error:
|
||||||
logger.warning("Blocked dangerous workdir: %s (command: %s)",
|
logger.warning("Blocked dangerous workdir: %s (command: %s)",
|
||||||
workdir[:200], command[:200])
|
workdir[:200], _safe_command_preview(command))
|
||||||
return json.dumps({
|
return json.dumps({
|
||||||
"output": "",
|
"output": "",
|
||||||
"exit_code": -1,
|
"exit_code": -1,
|
||||||
@@ -1347,12 +1374,12 @@ def terminal_tool(
|
|||||||
retry_count += 1
|
retry_count += 1
|
||||||
wait_time = 2 ** retry_count
|
wait_time = 2 ** retry_count
|
||||||
logger.warning("Execution error, retrying in %ds (attempt %d/%d) - Command: %s - Error: %s: %s - Task: %s, Backend: %s",
|
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)
|
time.sleep(wait_time)
|
||||||
continue
|
continue
|
||||||
|
|
||||||
logger.error("Execution failed after %d retries - Command: %s - Error: %s: %s - Task: %s, Backend: %s",
|
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({
|
return json.dumps({
|
||||||
"output": "",
|
"output": "",
|
||||||
"exit_code": -1,
|
"exit_code": -1,
|
||||||
|
|||||||
Reference in New Issue
Block a user