diff --git a/tests/tools/test_code_execution.py b/tests/tools/test_code_execution.py index 3521d19ea19..07dc188600c 100644 --- a/tests/tools/test_code_execution.py +++ b/tests/tools/test_code_execution.py @@ -174,6 +174,47 @@ class TestExecuteCodeRemoteTempDir(unittest.TestCase): self.assertIn("rm -rf /data/data/com.termux/files/usr/tmp/hermes_exec_", cleanup_cmd) self.assertNotIn("mkdir -p /tmp/hermes_exec_", mkdir_cmd) + def test_timezone_shell_quoted_in_remote_execution(self): + """HERMES_TIMEZONE must be shell-quoted in remote env_prefix to prevent injection.""" + class FakeEnv: + def __init__(self): + self.commands = [] + + def get_temp_dir(self): + return "/tmp" + + def execute(self, command, cwd=None, timeout=None): + self.commands.append((command, cwd, timeout)) + if "command -v python3" in command: + return {"output": "OK\n"} + if "python3 script.py" in command: + return {"output": "hello\n", "returncode": 0} + return {"output": ""} + + env = FakeEnv() + fake_thread = MagicMock() + + malicious_tz = "US/Eastern; echo PWNED" + + with patch("tools.code_execution_tool._load_config", + return_value={"timeout": 30, "max_tool_calls": 5}), \ + patch("tools.code_execution_tool._get_or_create_env", + return_value=(env, "ssh")), \ + patch("tools.code_execution_tool._ship_file_to_remote"), \ + patch("tools.code_execution_tool.threading.Thread", + return_value=fake_thread), \ + patch.dict(os.environ, {"HERMES_TIMEZONE": malicious_tz}): + result = json.loads(_execute_remote("print('hello')", "task-1", ["terminal"])) + + self.assertEqual(result["status"], "success") + run_cmd = next(cmd for cmd, _, _ in env.commands if "python3 script.py" in cmd) + # The TZ value must be shell-quoted — it should NOT contain unescaped semicolons + self.assertNotIn("TZ=US/Eastern; echo PWNED", run_cmd, + "TZ value with shell metacharacters must not appear unquoted") + # shlex.quote wraps values containing special characters in single quotes + self.assertIn("TZ='US/Eastern; echo PWNED'", run_cmd, + "TZ value must be wrapped in single quotes by shlex.quote()") + @unittest.skipIf(sys.platform == "win32", "UDS not available on Windows") class TestExecuteCode(unittest.TestCase): diff --git a/tools/code_execution_tool.py b/tools/code_execution_tool.py index 5514f63b9f7..5749b224bdf 100644 --- a/tools/code_execution_tool.py +++ b/tools/code_execution_tool.py @@ -961,7 +961,7 @@ def _execute_remote( ) tz = os.getenv("HERMES_TIMEZONE", "").strip() if tz: - env_prefix += f" TZ={tz}" + env_prefix += f" TZ={shlex.quote(tz)}" # Execute the script on the remote backend logger.info("Executing code on %s backend (task %s)...",