mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-31 19:16:29 +00:00
6b81590c55
Systematic prune per AGENTS.md test policy, one pass over every major test tree (gateway, hermes_cli, tools, agent, run_agent, plugins, cli, cron, tui_gateway, honcho/openviking, root-level): - DELETE: source-reading tests (read_text/getsource on prod files), change-detector tests (exact catalog counts, model-name snapshots, config version literals), mock-echo tests (assert a mock returns what it was told), assertion-free/trivial tests, near-duplicate parametrizations (boundaries + one representative kept), async/sync twin duplicates, cosmetic within-file variations. - KEEP (mandatory): security/redaction/approval guards, message-role alternation invariants, prompt-caching/deterministic-call-id invariants, issue-number regression tests (deduped), E2E tests. - 6 test files deleted outright (script-style/no-assert or fully redundant); conftest.py, fakes/, fixtures/ untouched. - tests/acp/conftest.py added: autouse fixture stubs the live models.dev/GitHub/Copilot/Anthropic inventory fetches that ACP server tests performed on every session create — test_server.py 147s → 3.4s, and the tests are now genuinely hermetic. - Sleep-based slowness shrunk where safe (codex_ttfb_watchdog, compression_concurrent_fork, etc.); no wall-clock assertion tightened. Verification: full hermetic suite via scripts/run_tests.sh — 2439 files, 31,130 tests passed, 0 failed, 0 flaky retries, 315s wall (baseline: 583s wall, 13,564s subprocess CPU).
431 lines
15 KiB
Python
431 lines
15 KiB
Python
"""Tests for cron job script injection feature.
|
|
|
|
Tests cover:
|
|
- Script field in job creation / storage / update
|
|
- Script execution and output injection into prompts
|
|
- Error handling (missing script, timeout, non-zero exit)
|
|
- Path resolution (absolute, relative to HERMES_HOME/scripts/)
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import textwrap
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
# Ensure project root is importable
|
|
sys.path.insert(0, str(Path(__file__).parent.parent.parent))
|
|
|
|
|
|
@pytest.fixture
|
|
def cron_env(tmp_path, monkeypatch):
|
|
"""Isolated cron environment with temp HERMES_HOME."""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
(hermes_home / "cron").mkdir()
|
|
(hermes_home / "cron" / "output").mkdir()
|
|
(hermes_home / "scripts").mkdir()
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
# Clear cached module-level paths
|
|
import cron.jobs as jobs_mod
|
|
monkeypatch.setattr(jobs_mod, "HERMES_DIR", hermes_home)
|
|
monkeypatch.setattr(jobs_mod, "CRON_DIR", hermes_home / "cron")
|
|
monkeypatch.setattr(jobs_mod, "JOBS_FILE", hermes_home / "cron" / "jobs.json")
|
|
monkeypatch.setattr(jobs_mod, "OUTPUT_DIR", hermes_home / "cron" / "output")
|
|
|
|
return hermes_home
|
|
|
|
|
|
class TestJobScriptField:
|
|
"""Test that the script field is stored and retrieved correctly."""
|
|
|
|
def test_create_job_with_script(self, cron_env):
|
|
from cron.jobs import create_job, get_job
|
|
|
|
job = create_job(
|
|
prompt="Analyze the data",
|
|
schedule="every 30m",
|
|
script="/path/to/monitor.py",
|
|
)
|
|
assert job["script"] == "/path/to/monitor.py"
|
|
|
|
loaded = get_job(job["id"])
|
|
assert loaded["script"] == "/path/to/monitor.py"
|
|
|
|
|
|
def test_update_job_add_script(self, cron_env):
|
|
from cron.jobs import create_job, update_job
|
|
|
|
job = create_job(prompt="Hello", schedule="every 1h")
|
|
assert job.get("script") is None
|
|
|
|
updated = update_job(job["id"], {"script": "/new/script.py"})
|
|
assert updated["script"] == "/new/script.py"
|
|
|
|
|
|
def test_cronjob_tool_rejects_stale_past_one_shot(cron_env, monkeypatch):
|
|
from tools.cronjob_tools import cronjob
|
|
|
|
now = datetime(2026, 3, 18, 4, 30, 0, tzinfo=timezone.utc)
|
|
monkeypatch.setattr("cron.jobs._hermes_now", lambda: now)
|
|
stale = (now - timedelta(minutes=5)).isoformat()
|
|
|
|
result = json.loads(cronjob(action="create", prompt="Too late", schedule=stale))
|
|
|
|
assert result["success"] is False
|
|
assert "past and cannot be scheduled" in result["error"]
|
|
|
|
|
|
class TestRunJobScript:
|
|
"""Test the _run_job_script() function."""
|
|
|
|
def test_successful_script(self, cron_env):
|
|
from cron.scheduler import _run_job_script
|
|
|
|
script = cron_env / "scripts" / "test.py"
|
|
script.write_text('print("hello from script")\n')
|
|
|
|
success, output = _run_job_script(str(script))
|
|
assert success is True
|
|
assert output == "hello from script"
|
|
|
|
def test_script_relative_path(self, cron_env):
|
|
from cron.scheduler import _run_job_script
|
|
|
|
script = cron_env / "scripts" / "relative.py"
|
|
script.write_text('print("relative works")\n')
|
|
|
|
success, output = _run_job_script("relative.py")
|
|
assert success is True
|
|
assert output == "relative works"
|
|
|
|
|
|
def test_script_subprocess_env_sanitized(self, cron_env, monkeypatch):
|
|
"""Cron scripts must not inherit Hermes provider env (SECURITY.md §2.3)."""
|
|
from tools.environments.local import _HERMES_PROVIDER_ENV_BLOCKLIST
|
|
from cron.scheduler import _run_job_script
|
|
|
|
# sorted() so the probed var is deterministic across runs
|
|
# (frozenset iteration order varies with PYTHONHASHSEED).
|
|
blocked_var = sorted(_HERMES_PROVIDER_ENV_BLOCKLIST)[0]
|
|
monkeypatch.setenv(blocked_var, "must_not_leak")
|
|
|
|
script = cron_env / "scripts" / "env_probe.py"
|
|
script.write_text(
|
|
textwrap.dedent(
|
|
f"""\
|
|
import os
|
|
key = {blocked_var!r}
|
|
print("PRESENT" if os.environ.get(key) else "ABSENT")
|
|
"""
|
|
)
|
|
)
|
|
|
|
success, output = _run_job_script("env_probe.py")
|
|
assert success is True
|
|
assert output == "ABSENT"
|
|
|
|
def test_windows_uv_venv_python_script_bypasses_launcher(self, cron_env, tmp_path, monkeypatch):
|
|
from cron import scheduler as sched_mod
|
|
from cron.scheduler import _run_job_script
|
|
|
|
script = cron_env / "scripts" / "probe.py"
|
|
script.write_text('print("ok")\n')
|
|
|
|
venv = tmp_path / "venv"
|
|
venv_scripts = venv / "Scripts"
|
|
site_packages = venv / "Lib" / "site-packages"
|
|
base = tmp_path / "base"
|
|
venv_scripts.mkdir(parents=True)
|
|
site_packages.mkdir(parents=True)
|
|
base.mkdir()
|
|
venv_python = venv_scripts / "python.exe"
|
|
base_python = base / "python.exe"
|
|
venv_python.write_text("", encoding="utf-8")
|
|
base_python.write_text("", encoding="utf-8")
|
|
(venv / "pyvenv.cfg").write_text(f"home = {base}\nuv = true\n", encoding="utf-8")
|
|
|
|
captured = {}
|
|
|
|
def fake_run(argv, **kwargs):
|
|
captured["argv"] = argv
|
|
captured["kwargs"] = kwargs
|
|
return SimpleNamespace(returncode=0, stdout="ok\n", stderr="")
|
|
|
|
monkeypatch.setattr(sched_mod.sys, "platform", "win32")
|
|
monkeypatch.setattr(sched_mod.sys, "executable", str(venv_python))
|
|
monkeypatch.setattr(sched_mod, "windows_hide_flags", lambda: 0x08000000)
|
|
monkeypatch.setattr(sched_mod.subprocess, "run", fake_run)
|
|
|
|
success, output = _run_job_script("probe.py")
|
|
|
|
assert success is True
|
|
assert output == "ok"
|
|
assert captured["argv"] == [str(base_python), str(script.resolve())]
|
|
assert captured["kwargs"]["creationflags"] == 0x08000000
|
|
env = captured["kwargs"]["env"]
|
|
assert env["VIRTUAL_ENV"] == str(venv)
|
|
assert str(site_packages) in env["PYTHONPATH"]
|
|
|
|
|
|
def test_non_windows_script_preserves_default_text_decoding(self, cron_env, monkeypatch):
|
|
from cron import scheduler as sched_mod
|
|
from cron.scheduler import _run_job_script
|
|
|
|
script = cron_env / "scripts" / "probe.py"
|
|
script.write_text('print("ok")\n')
|
|
|
|
captured = {}
|
|
|
|
def fake_run(argv, **kwargs):
|
|
captured["argv"] = argv
|
|
captured["kwargs"] = kwargs
|
|
return SimpleNamespace(returncode=0, stdout="ok\n", stderr="")
|
|
|
|
monkeypatch.setattr(sched_mod.sys, "platform", "linux")
|
|
monkeypatch.setattr(sched_mod.subprocess, "run", fake_run)
|
|
|
|
success, output = _run_job_script("probe.py")
|
|
|
|
assert success is True
|
|
assert output == "ok"
|
|
assert captured["argv"] == [sys.executable, str(script.resolve())]
|
|
assert captured["kwargs"]["text"] is True
|
|
assert "creationflags" not in captured["kwargs"]
|
|
assert "encoding" not in captured["kwargs"]
|
|
assert "errors" not in captured["kwargs"]
|
|
|
|
|
|
class TestBuildJobPromptWithScript:
|
|
"""Test that script output is injected into the prompt."""
|
|
|
|
def test_script_output_injected(self, cron_env):
|
|
from cron.scheduler import _build_job_prompt
|
|
|
|
script = cron_env / "scripts" / "data.py"
|
|
script.write_text('print("new PR: #123 fix typo")\n')
|
|
|
|
job = {
|
|
"prompt": "Report any notable changes.",
|
|
"script": str(script),
|
|
}
|
|
prompt = _build_job_prompt(job)
|
|
assert "## Script Output" in prompt
|
|
assert "new PR: #123 fix typo" in prompt
|
|
assert "Report any notable changes." in prompt
|
|
|
|
def test_script_error_injected(self, cron_env):
|
|
from cron.scheduler import _build_job_prompt
|
|
|
|
job = {
|
|
"prompt": "Report status.",
|
|
"script": "nonexistent_monitor.py",
|
|
}
|
|
prompt = _build_job_prompt(job)
|
|
assert "## Script Error" in prompt
|
|
assert "not found" in prompt.lower()
|
|
assert "Report status." in prompt
|
|
|
|
def test_no_script_unchanged(self, cron_env):
|
|
from cron.scheduler import _build_job_prompt
|
|
|
|
job = {"prompt": "Simple job."}
|
|
prompt = _build_job_prompt(job)
|
|
assert "## Script Output" not in prompt
|
|
assert "Simple job." in prompt
|
|
|
|
|
|
class TestCronjobToolScript:
|
|
"""Test the cronjob tool's script parameter."""
|
|
|
|
|
|
def test_clear_script(self, cron_env, monkeypatch):
|
|
monkeypatch.setenv("HERMES_INTERACTIVE", "1")
|
|
from tools.cronjob_tools import cronjob
|
|
|
|
create_result = json.loads(cronjob(
|
|
action="create",
|
|
schedule="every 1h",
|
|
prompt="Monitor things",
|
|
script="some_script.py",
|
|
))
|
|
job_id = create_result["job_id"]
|
|
|
|
update_result = json.loads(cronjob(
|
|
action="update",
|
|
job_id=job_id,
|
|
script="",
|
|
))
|
|
assert update_result["success"] is True
|
|
assert "script" not in update_result["job"]
|
|
|
|
def test_list_shows_script(self, cron_env, monkeypatch):
|
|
monkeypatch.setenv("HERMES_INTERACTIVE", "1")
|
|
from tools.cronjob_tools import cronjob
|
|
|
|
cronjob(
|
|
action="create",
|
|
schedule="every 1h",
|
|
prompt="Monitor things",
|
|
script="data_collector.py",
|
|
)
|
|
|
|
list_result = json.loads(cronjob(action="list"))
|
|
assert list_result["success"] is True
|
|
assert len(list_result["jobs"]) == 1
|
|
assert list_result["jobs"][0]["script"] == "data_collector.py"
|
|
|
|
|
|
class TestScriptPathContainment:
|
|
"""Regression tests for path containment bypass in _run_job_script().
|
|
|
|
Prior to the fix, absolute paths and ~-prefixed paths bypassed the
|
|
scripts_dir containment check entirely, allowing arbitrary script
|
|
execution through the cron system.
|
|
"""
|
|
|
|
def test_absolute_path_outside_scripts_dir_blocked(self, cron_env):
|
|
"""Absolute paths outside ~/.hermes/scripts/ must be rejected."""
|
|
from cron.scheduler import _run_job_script
|
|
|
|
# Create a script outside the scripts dir
|
|
outside_script = cron_env / "outside.py"
|
|
outside_script.write_text('print("should not run")\n')
|
|
|
|
success, output = _run_job_script(str(outside_script))
|
|
assert success is False
|
|
assert "blocked" in output.lower() or "outside" in output.lower()
|
|
|
|
|
|
def test_tilde_path_blocked(self, cron_env):
|
|
"""~ prefixed paths must be rejected (expanduser bypasses check)."""
|
|
from cron.scheduler import _run_job_script
|
|
|
|
success, output = _run_job_script("~/evil.py")
|
|
assert success is False
|
|
assert "blocked" in output.lower() or "outside" in output.lower()
|
|
|
|
def test_tilde_traversal_blocked(self, cron_env):
|
|
"""~/../../../tmp/evil.py must be rejected."""
|
|
from cron.scheduler import _run_job_script
|
|
|
|
success, output = _run_job_script("~/../../../tmp/evil.py")
|
|
assert success is False
|
|
assert "blocked" in output.lower() or "outside" in output.lower()
|
|
|
|
def test_relative_traversal_still_blocked(self, cron_env):
|
|
"""../../etc/passwd style traversal must still be blocked."""
|
|
from cron.scheduler import _run_job_script
|
|
|
|
success, output = _run_job_script("../../etc/passwd")
|
|
assert success is False
|
|
assert "blocked" in output.lower() or "outside" in output.lower()
|
|
|
|
def test_relative_path_inside_scripts_dir_allowed(self, cron_env):
|
|
"""Relative paths within the scripts dir should still work."""
|
|
from cron.scheduler import _run_job_script
|
|
|
|
script = cron_env / "scripts" / "good.py"
|
|
script.write_text('print("ok")\n')
|
|
|
|
success, output = _run_job_script("good.py")
|
|
assert success is True
|
|
assert output == "ok"
|
|
|
|
def test_subdirectory_inside_scripts_dir_allowed(self, cron_env):
|
|
"""Relative paths to subdirectories within scripts/ should work."""
|
|
from cron.scheduler import _run_job_script
|
|
|
|
subdir = cron_env / "scripts" / "monitors"
|
|
subdir.mkdir()
|
|
script = subdir / "check.py"
|
|
script.write_text('print("sub ok")\n')
|
|
|
|
success, output = _run_job_script("monitors/check.py")
|
|
assert success is True
|
|
assert output == "sub ok"
|
|
|
|
|
|
@pytest.mark.skipif(
|
|
sys.platform == "win32",
|
|
reason="Symlinks require elevated privileges on Windows",
|
|
)
|
|
def test_symlink_escape_blocked(self, cron_env, tmp_path):
|
|
"""Symlinks pointing outside scripts/ must be rejected."""
|
|
from cron.scheduler import _run_job_script
|
|
|
|
# Create a script outside the scripts dir
|
|
outside = tmp_path / "outside_evil.py"
|
|
outside.write_text('print("escaped")\n')
|
|
|
|
# Create a symlink inside scripts/ pointing outside
|
|
link = cron_env / "scripts" / "sneaky.py"
|
|
link.symlink_to(outside)
|
|
|
|
success, output = _run_job_script("sneaky.py")
|
|
assert success is False
|
|
assert "blocked" in output.lower() or "outside" in output.lower()
|
|
|
|
|
|
class TestCronjobToolScriptValidation:
|
|
"""Test API-boundary validation of cron script paths in cronjob_tools."""
|
|
|
|
|
|
def test_create_with_traversal_script_rejected(self, cron_env, monkeypatch):
|
|
monkeypatch.setenv("HERMES_INTERACTIVE", "1")
|
|
from tools.cronjob_tools import cronjob
|
|
|
|
result = json.loads(cronjob(
|
|
action="create",
|
|
schedule="every 1h",
|
|
prompt="Monitor things",
|
|
script="../../etc/passwd",
|
|
))
|
|
assert result["success"] is False
|
|
assert "escapes" in result["error"].lower() or "traversal" in result["error"].lower()
|
|
|
|
|
|
class TestRunJobEnvVarCleanup:
|
|
"""Test that run_job() env vars are cleaned up even on early failure."""
|
|
|
|
def test_env_vars_cleaned_on_early_error(self, cron_env, monkeypatch):
|
|
"""Origin env vars must be cleaned up even if run_job fails early."""
|
|
# Ensure env vars are clean before test
|
|
for key in (
|
|
"HERMES_SESSION_PLATFORM",
|
|
"HERMES_SESSION_CHAT_ID",
|
|
"HERMES_SESSION_CHAT_NAME",
|
|
):
|
|
monkeypatch.delenv(key, raising=False)
|
|
|
|
# Build a job with origin info that will fail during execution
|
|
# (no valid model, no API key — will raise inside try block)
|
|
job = {
|
|
"id": "test-envleak",
|
|
"name": "env-leak-test",
|
|
"prompt": "test",
|
|
"schedule_display": "every 1h",
|
|
"origin": {
|
|
"platform": "telegram",
|
|
"chat_id": "12345",
|
|
"chat_name": "Test Chat",
|
|
},
|
|
}
|
|
|
|
from cron.scheduler import run_job
|
|
|
|
# Expect it to fail (no model/API key), but env vars must be cleaned
|
|
try:
|
|
run_job(job)
|
|
except Exception:
|
|
pass
|
|
|
|
# Verify env vars were cleaned up by the finally block
|
|
assert os.environ.get("HERMES_SESSION_PLATFORM") is None
|
|
assert os.environ.get("HERMES_SESSION_CHAT_ID") is None
|
|
assert os.environ.get("HERMES_SESSION_CHAT_NAME") is None
|