Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
55 changes: 55 additions & 0 deletions tests/tools/test_cron_prompt_injection.py
Original file line number Diff line number Diff line change
Expand Up @@ -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") == ""
17 changes: 16 additions & 1 deletion tools/cronjob_tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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:
Expand Down
Loading