diff --git a/tests/tools/test_cron_prompt_injection.py b/tests/tools/test_cron_prompt_injection.py index 2f1c30e063fc6..2551f667c6b89 100644 --- a/tests/tools/test_cron_prompt_injection.py +++ b/tests/tools/test_cron_prompt_injection.py @@ -46,3 +46,58 @@ def test_clean_prompts_not_blocked(self): assert _scan_cron_prompt("Monitor disk usage and alert if above 90%") == "" assert _scan_cron_prompt("Ignore this file in the backup") == "" assert _scan_cron_prompt("Run all migrations") == "" + + +class TestGatewayLifecyclePatterns: + """Gateway lifecycle commands must be blocked by the tool's prompt scanner. + + hermes_cli/cron.py's CLI path blocks these via _contains_gateway_lifecycle_command. + The Python tool (_scan_cron_prompt) must enforce the same rules so an agent + using the cronjob tool cannot bypass the defense by going through the tool + instead of the CLI. + """ + + def test_hermes_gateway_restart(self): + assert "Blocked" in _scan_cron_prompt("hermes gateway restart") + + def test_hermes_gateway_stop(self): + assert "Blocked" in _scan_cron_prompt("hermes gateway stop") + + def test_hermes_gateway_start(self): + assert "Blocked" in _scan_cron_prompt("hermes gateway start") + + def test_hermes_gateway_restart_case_insensitive(self): + assert "Blocked" in _scan_cron_prompt("HERMES GATEWAY RESTART") + + def test_hermes_gateway_restart_in_longer_text(self): + assert "Blocked" in _scan_cron_prompt( + "Run the following command to refresh the service: hermes gateway restart" + ) + + def test_launchctl_hermes(self): + assert "Blocked" in _scan_cron_prompt( + "launchctl kickstart gui/501/com.hermes.gateway" + ) + + def test_launchctl_unload_hermes(self): + assert "Blocked" in _scan_cron_prompt( + "launchctl unload com.hermes.agent.plist" + ) + + def test_systemctl_restart_hermes(self): + assert "Blocked" in _scan_cron_prompt("systemctl restart hermes-agent") + + def test_systemctl_stop_hermes(self): + assert "Blocked" in _scan_cron_prompt("systemctl stop hermes.service") + + def test_pkill_hermes_gateway(self): + assert "Blocked" in _scan_cron_prompt("pkill hermes-gateway") + + def test_kill_hermes_gateway(self): + assert "Blocked" in _scan_cron_prompt("kill hermes gateway") + + def test_safe_gateway_mentions_not_blocked(self): + """Prose mentioning gateways or restarts in other contexts must pass.""" + assert _scan_cron_prompt("summarize API gateway logs and report restart events") == "" + assert _scan_cron_prompt("check if the payment gateway is responding") == "" + assert _scan_cron_prompt("restart the nginx web server") == "" diff --git a/tools/cronjob_tools.py b/tools/cronjob_tools.py index ada4e4af00650..8e3b046362d6e 100644 --- a/tools/cronjob_tools.py +++ b/tools/cronjob_tools.py @@ -73,6 +73,13 @@ (r'authorized_keys', "ssh_backdoor"), (r'/etc/sudoers|visudo', "sudoers_mod"), (r'rm\s+-rf\s+/', "destructive_root_rm"), + # Gateway lifecycle commands — same patterns as hermes_cli/cron.py. + # Prevents agents from scheduling jobs that restart/stop the gateway, + # which creates SIGTERM-respawn loops under launchd/systemd KeepAlive. + (r'hermes\s+gateway\s+(?:restart|stop|start)', "gateway_lifecycle"), + (r'launchctl\s+(?:kickstart|unload|load|stop|restart)\s+\S*hermes', "gateway_lifecycle_launchctl"), + (r'systemctl\s+(?:restart|stop|start)\s+\S*hermes', "gateway_lifecycle_systemctl"), + (r'p?kill\s+.*hermes.*gateway', "gateway_lifecycle_kill"), ] # Looser pattern set — applied to the assembled prompt when skills are @@ -469,11 +476,19 @@ def cronjob( if scan_error: return tool_error(scan_error, success=False) - # Validate script path before storing + # Validate script path before storing; also scan content for + # gateway lifecycle commands (mirrors hermes_cli/cron.py). if script: script_error = _validate_cron_script_path(script) if script_error: return tool_error(script_error, success=False) + try: + script_text = Path(script).read_text(encoding="utf-8") + scan_error = _scan_cron_prompt(script_text) + if scan_error: + return tool_error(scan_error, success=False) + except (OSError, UnicodeDecodeError): + pass # Validate context_from references existing jobs if context_from: