Harden cron job output paths

_docs/2026-05-21_cron-output-path-hardening_codex.md

@@ -0,0 +1,46 @@
+# Cron Output Path Hardening
+
+Date: 2026-05-21
+Branch: codex/hermes-cron-path-hardening-20260521
+
+## Summary
+
+Hardened cron job update and output handling so dashboard/API updates cannot
+move a job's immutable `id` into a filesystem path, and output writes/deletes
+require a single safe output directory component.
+
+Changes:
+
+- `cron.jobs.update_job()` rejects immutable `id` updates.
+- `cron.jobs.save_job_output()` resolves output directories through a shared
+  helper that rejects absolute paths, parent traversal, and nested components.
+- `cron.jobs.remove_job()` uses the same helper before deleting output
+  directories.
+- Dashboard cron update/delete endpoints convert these validation failures into
+  HTTP 400 responses.
+
+## Verification
+
+Commands run:
+
+```powershell
+$env:UV_PROJECT_ENVIRONMENT = Join-Path $env:TEMP 'hermes-agent-codex-test-env'
+uv run --extra dev python -m pytest tests\cron\test_jobs.py::TestJobCRUD::test_remove_job_rejects_unsafe_legacy_id_before_output_cleanup tests\cron\test_jobs.py::TestUpdateJob::test_update_rejects_id_change tests\cron\test_jobs.py::TestSaveJobOutput -q --timeout-method=thread
+uv run --extra dev --extra web python -m pytest tests\hermes_cli\test_web_server_cron_profiles.py::test_update_cron_job_rejects_id_mutation -q --timeout-method=thread
+uv run --extra dev python -m pytest tests\cron\test_jobs.py -q --timeout-method=thread
+uv run --extra dev --extra web python -m pytest tests\hermes_cli\test_web_server_cron_profiles.py -q --timeout-method=thread
+uv run --extra dev python -m compileall cron\jobs.py hermes_cli\web_server.py
+git diff --check
+```
+
+Results:
+
+- Focused cron core tests: `7 passed`.
+- Focused dashboard cron update test: `1 passed`.
+- `tests\cron\test_jobs.py`: `85 passed`.
+- `tests\hermes_cli\test_web_server_cron_profiles.py`: `7 passed`.
+- `compileall` completed successfully.
+- `git diff --check` reported no whitespace errors.
+
+Note: the checkout-local `.venv` still lacks `pytest`; verification used a
+temporary `UV_PROJECT_ENVIRONMENT`.

cron/jobs.py

@@ -44,6 +44,34 @@
 _jobs_file_lock = threading.Lock()
 OUTPUT_DIR = CRON_DIR / "output"
 ONESHOT_GRACE_SECONDS = 120
+IMMUTABLE_UPDATE_FIELDS = {"id"}
+
+
+def _validate_job_output_id(job_id: str) -> str:
+    """Return a single safe path component for cron output directories."""
+    text = str(job_id or "").strip()
+    path = Path(text)
+    if (
+        not text
+        or path.is_absolute()
+        or path.drive
+        or len(path.parts) != 1
+        or path.parts[0] in {".", ".."}
+    ):
+        raise ValueError("Invalid cron job id for output path")
+    return text
+
+
+def _job_output_dir(job_id: str) -> Path:
+    """Resolve a job output directory and require it to stay under OUTPUT_DIR."""
+    safe_id = _validate_job_output_id(job_id)
+    base = OUTPUT_DIR.resolve(strict=False)
+    target = (base / safe_id).resolve(strict=False)
+    try:
+        target.relative_to(base)
+    except ValueError as exc:
+        raise ValueError("Invalid cron job id for output path") from exc
+    return target
 
 
 def _normalize_skill_list(skill: Optional[str] = None, skills: Optional[Any] = None) -> List[str]:
@@ -728,6 +756,12 @@ def list_jobs(include_disabled: bool = False) -> List[Dict[str, Any]]:
 
 def update_job(job_id: str, updates: Dict[str, Any]) -> Optional[Dict[str, Any]]:
     """Update a job by ID, refreshing derived schedule fields when needed."""
+    updates = dict(updates or {})
+    immutable_updates = IMMUTABLE_UPDATE_FIELDS.intersection(updates)
+    if immutable_updates:
+        fields = ", ".join(sorted(immutable_updates))
+        raise ValueError(f"Cron job field(s) cannot be updated: {fields}")
+
     jobs = load_jobs()
     for i, job in enumerate(jobs):
         if job["id"] != job_id:
@@ -845,9 +879,9 @@ def remove_job(job_id: str) -> bool:
     original_len = len(jobs)
     jobs = [j for j in jobs if j["id"] != canonical_id]
     if len(jobs) < original_len:
+        job_output_dir = _job_output_dir(canonical_id)
         save_jobs(jobs)
         # Clean up output directory to prevent orphaned dirs accumulating
-        job_output_dir = OUTPUT_DIR / canonical_id
         if job_output_dir.exists():
             shutil.rmtree(job_output_dir)
         return True
@@ -1061,7 +1095,7 @@ def _get_due_jobs_locked() -> List[Dict[str, Any]]:
 def save_job_output(job_id: str, output: str):
     """Save job output to file."""
     ensure_dirs()
-    job_output_dir = OUTPUT_DIR / job_id
+    job_output_dir = _job_output_dir(job_id)
     job_output_dir.mkdir(parents=True, exist_ok=True)
     _secure_dir(job_output_dir)
 

hermes_cli/web_server.py

@@ -2690,7 +2690,10 @@ async def update_cron_job(job_id: str, body: CronJobUpdate, profile: Optional[st
     selected = profile or _find_cron_job_profile(job_id)
     if not selected:
         raise HTTPException(status_code=404, detail="Job not found")
-    job = _call_cron_for_profile(selected, "update_job", job_id, body.updates)
+    try:
+        job = _call_cron_for_profile(selected, "update_job", job_id, body.updates)
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
     if not job:
         raise HTTPException(status_code=404, detail="Job not found")
     return job
@@ -2734,7 +2737,11 @@ async def delete_cron_job(job_id: str, profile: Optional[str] = None):
     selected = profile or _find_cron_job_profile(job_id)
     if not selected:
         raise HTTPException(status_code=404, detail="Job not found")
-    if not _call_cron_for_profile(selected, "remove_job", job_id):
+    try:
+        removed = _call_cron_for_profile(selected, "remove_job", job_id)
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+    if not removed:
         raise HTTPException(status_code=404, detail="Job not found")
     return {"ok": True}
 

tests/cron/test_jobs.py

@@ -232,6 +232,20 @@ def test_remove_job(self, tmp_cron_dir):
         assert remove_job(job["id"]) is True
         assert get_job(job["id"]) is None
 
+    def test_remove_job_rejects_unsafe_legacy_id_before_output_cleanup(self, tmp_cron_dir):
+        job = create_job(prompt="Legacy unsafe", schedule="every 1h")
+        job["id"] = "../escape"
+        save_jobs([job])
+        outside = tmp_cron_dir / "escape"
+        outside.mkdir()
+        (outside / "keep.txt").write_text("keep", encoding="utf-8")
+
+        with pytest.raises(ValueError, match="output path"):
+            remove_job("../escape")
+
+        assert load_jobs()[0]["id"] == "../escape"
+        assert (outside / "keep.txt").exists()
+
     def test_remove_nonexistent_returns_false(self, tmp_cron_dir):
         assert remove_job("nonexistent") is False
 
@@ -300,6 +314,15 @@ def test_update_nonexistent_returns_none(self, tmp_cron_dir):
         result = update_job("nonexistent_id", {"name": "X"})
         assert result is None
 
+    def test_update_rejects_id_change(self, tmp_cron_dir):
+        job = create_job(prompt="Original", schedule="every 1h")
+
+        with pytest.raises(ValueError, match="id"):
+            update_job(job["id"], {"id": "../escape"})
+
+        assert get_job(job["id"]) is not None
+        assert get_job("../escape") is None
+
 
 class TestPauseResumeJob:
     def test_pause_sets_state(self, tmp_cron_dir):
@@ -953,3 +976,16 @@ def test_creates_output_file(self, tmp_cron_dir):
         assert output_file.exists()
         assert output_file.read_text() == "# Results\nEverything ok."
         assert "test123" in str(output_file)
+
+    @pytest.mark.parametrize("bad_job_id", ["../escape", "nested/escape", "."])
+    def test_rejects_unsafe_job_id(self, tmp_cron_dir, bad_job_id):
+        with pytest.raises(ValueError, match="output path"):
+            save_job_output(bad_job_id, "# Results")
+
+        assert not (tmp_cron_dir / "escape").exists()
+
+    def test_rejects_absolute_job_id(self, tmp_cron_dir):
+        with pytest.raises(ValueError, match="output path"):
+            save_job_output(str(tmp_cron_dir / "outside"), "# Results")
+
+        assert not (tmp_cron_dir / "outside").exists()

tests/hermes_cli/test_web_server_cron_profiles.py

@@ -131,6 +131,31 @@ async def test_cron_mutation_without_profile_finds_named_profile_job(isolated_pr
     assert worker_jobs[0]["enabled"] is False
 
 
+@pytest.mark.asyncio
+async def test_update_cron_job_rejects_id_mutation(isolated_profiles):
+    from hermes_cli import web_server
+
+    worker_job = web_server._call_cron_for_profile(
+        "worker_alpha",
+        "create_job",
+        prompt="managed by named profile",
+        schedule="every 1h",
+        name="immutable-id-job",
+    )
+
+    with pytest.raises(HTTPException) as exc:
+        await web_server.update_cron_job(
+            worker_job["id"],
+            web_server.CronJobUpdate(updates={"id": "../escape"}),
+            profile="worker_alpha",
+        )
+
+    assert exc.value.status_code == 400
+    assert "id" in exc.value.detail
+    worker_jobs = await web_server.list_cron_jobs(profile="worker_alpha")
+    assert [job["id"] for job in worker_jobs] == [worker_job["id"]]
+
+
 @pytest.mark.asyncio
 async def test_cron_delete_with_profile_deletes_only_target_profile(isolated_profiles):
     from hermes_cli import web_server