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
47 changes: 45 additions & 2 deletions cron/jobs.py
Original file line number Diff line number Diff line change
Expand Up @@ -471,6 +471,44 @@ def save_jobs(jobs: List[Dict[str, Any]]):
raise


def _normalize_script_path(script: Optional[Any]) -> Optional[str]:
"""Normalize a cron job script reference.

The scheduler resolves relative paths against ``HERMES_HOME/scripts/``
(see ``cron.scheduler._run_job_script``). A user-supplied
``"scripts/foo.py"`` would therefore resolve to
``HERMES_HOME/scripts/scripts/foo.py`` and silently fall through the
"Script not found" branch at runtime — the script appears to vanish
between configuration and execution. Strip a leading ``scripts``
segment from relative paths to short-circuit that trap at create /
update time.

Absolute paths and ``~``-prefixed paths are returned verbatim — the
scheduler's existing path-traversal guard validates them against the
scripts directory at runtime, and rewriting them here would risk
silently re-pointing a user's explicit absolute path.
"""
if not isinstance(script, str):
return None
raw = script.strip()
if not raw:
return None
candidate = Path(raw)
if candidate.is_absolute() or raw.startswith("~"):
return raw
parts = candidate.parts
if len(parts) > 1 and parts[0] == "scripts":
stripped = str(Path(*parts[1:]))
logger.info(
"Cron script path %r normalized to %r — relative paths resolve "
"under HERMES_HOME/scripts/, so a leading 'scripts/' prefix "
"would create a scripts/scripts/ double-directory trap.",
raw, stripped,
)
return stripped
return raw


def _normalize_workdir(workdir: Optional[str]) -> Optional[str]:
"""Normalize and validate a cron job workdir.

Expand Down Expand Up @@ -623,8 +661,7 @@ def create_job(
normalized_model = normalized_model or None
normalized_provider = normalized_provider or None
normalized_base_url = normalized_base_url or None
normalized_script = str(script).strip() if isinstance(script, str) else None
normalized_script = normalized_script or None
normalized_script = _normalize_script_path(script)
normalized_toolsets = [str(t).strip() for t in enabled_toolsets if str(t).strip()] if enabled_toolsets else None
normalized_toolsets = normalized_toolsets or None
normalized_workdir = _normalize_workdir(workdir)
Expand Down Expand Up @@ -782,6 +819,12 @@ def update_job(job_id: str, updates: Dict[str, Any]) -> Optional[Dict[str, Any]]
else:
updates["profile"] = _normalize_profile(_profile)

# Normalize script if present in updates — same scripts/ prefix
# stripping create_job applies, so users can switch a job's script
# via update without re-introducing the double-directory trap.
if "script" in updates:
updates["script"] = _normalize_script_path(updates["script"])

updated = _apply_skill_fields({**job, **updates})
schedule_changed = "schedule" in updates

Expand Down
107 changes: 107 additions & 0 deletions tests/cron/test_cron_script.py
Original file line number Diff line number Diff line change
Expand Up @@ -86,6 +86,113 @@ def test_update_job_clear_script(self, cron_env):
assert updated.get("script") is None


class TestScriptPathNormalization:
"""Regression guard for issue #26595 — leading ``scripts/`` prefix.

The scheduler resolves relative paths against ``HERMES_HOME/scripts/``,
so a user-supplied ``scripts/foo.py`` would otherwise resolve to
``HERMES_HOME/scripts/scripts/foo.py`` and silently fail at runtime.
"""

def test_create_job_strips_leading_scripts_prefix(self, cron_env):
from cron.jobs import create_job

job = create_job(
prompt="Hello",
schedule="every 1h",
script="scripts/monitor.py",
)
assert job["script"] == "monitor.py"

def test_create_job_strips_leading_scripts_prefix_for_nested_path(self, cron_env):
from cron.jobs import create_job

job = create_job(
prompt="Hello",
schedule="every 1h",
script="scripts/sub/monitor.py",
)
assert job["script"] == str(Path("sub") / "monitor.py")

def test_create_job_preserves_unprefixed_relative_path(self, cron_env):
from cron.jobs import create_job

job = create_job(
prompt="Hello",
schedule="every 1h",
script="monitor.py",
)
assert job["script"] == "monitor.py"

def test_create_job_preserves_absolute_path(self, cron_env):
from cron.jobs import create_job

job = create_job(
prompt="Hello",
schedule="every 1h",
script="/abs/path/scripts/monitor.py",
)
assert job["script"] == "/abs/path/scripts/monitor.py"

def test_create_job_preserves_tilde_prefixed_path(self, cron_env):
from cron.jobs import create_job

job = create_job(
prompt="Hello",
schedule="every 1h",
script="~/scripts/monitor.py",
)
assert job["script"] == "~/scripts/monitor.py"

def test_create_job_preserves_lone_scripts_segment(self, cron_env):
"""A relative path of just ``scripts`` (no trailing segment) is left
alone — there is nothing to strip, and treating it as a script-name
could mask a genuine misconfiguration at runtime."""
from cron.jobs import create_job

job = create_job(
prompt="Hello",
schedule="every 1h",
script="scripts",
)
assert job["script"] == "scripts"

def test_update_job_strips_leading_scripts_prefix(self, cron_env):
from cron.jobs import create_job, update_job

job = create_job(prompt="Hello", schedule="every 1h")
updated = update_job(job["id"], {"script": "scripts/foo.py"})
assert updated["script"] == "foo.py"

def test_update_job_clears_script_with_empty_string(self, cron_env):
from cron.jobs import create_job, update_job

job = create_job(prompt="Hello", schedule="every 1h", script="monitor.py")
updated = update_job(job["id"], {"script": " "})
assert updated.get("script") is None

def test_normalized_script_resolves_to_real_file_at_runtime(self, cron_env):
"""End-to-end regression guard: a job created with ``scripts/foo.py``
must actually execute when the scheduler resolves its script path.
Before the fix, the runtime resolution path was
``HERMES_HOME/scripts/scripts/foo.py`` and the script ran nothing."""
from cron.jobs import create_job
from cron.scheduler import _run_job_script

script = cron_env / "scripts" / "monitor.py"
script.write_text('print("ran via normalized path")\n')

job = create_job(
prompt="Analyze",
schedule="every 30m",
script="scripts/monitor.py",
)

success, output = _run_job_script(job["script"])
assert success is True
assert output == "ran via normalized path"


class TestRunJobScript:
"""Test the _run_job_script() function."""

Expand Down
Loading