Kanban worker permissions and task-ownership controls

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #19713 fix(kanban): enforce worker task-ownership on destructive tool calls

  merged to main

  files: tests/tools/test_kanban_tools.py, tools/kanban_tools.py

  Show the diff

  • tests/tools/test_kanban_tools.py

    @@ -492,3 +492,121 @@ def test_kanban_guidance_prompt_size_bounded(monkeypatch, tmp_path):
         assert 1_500 < len(KANBAN_GUIDANCE) < 4_096, (
             f"KANBAN_GUIDANCE is {len(KANBAN_GUIDANCE)} chars — too short (missing?) or too long"
         )
    +
    +
    +# ---------------------------------------------------------------------------
    +# Worker task-ownership enforcement (regression tests for #19534)
    +# ---------------------------------------------------------------------------
    +#
    +# A worker process has HERMES_KANBAN_TASK set to its own task id. The
    +# destructive tools (kanban_complete, kanban_block, kanban_heartbeat)
    +# must refuse to operate on any OTHER task id, even if the caller
    +# supplies an explicit `task_id` argument. Workers legitimately call
    +# kanban_show / kanban_comment / kanban_create / kanban_link on other
    +# tasks, so those are unrestricted.
    +#
    +# Orchestrator profiles (no HERMES_KANBAN_TASK in env) are intentionally
    +# exempt — their job is routing, and they sometimes close out child
    +# tasks on behalf of the child.
    +
    +
    +def test_worker_complete_rejects_foreign_task_id(worker_env):
    +    """A worker cannot complete a task that isn't its own (#19534)."""
    +    from hermes_cli import kanban_db as kb
    +    conn = kb.connect()
    +    try:
    +        other = kb.create_task(conn, title="sibling")
    +        conn.execute("UPDATE tasks SET status='ready' WHERE id=?", (other,))
    +        conn.commit()
    +    finally:
    +        conn.close()
    +
    +    from tools import kanban_tools as kt
    +    out = kt._handle_complete({"task_id": other, "summary": "HIJACK"})
    +    d = json.loads(out)
    +    assert d.get("ok") is not True
    +    assert "refusing to mutate" in d.get("error", "")
    +
    +    # Sibling task must be untouched.
    +    conn = kb.connect()
    +    try:
    +        assert kb.get_task(conn, other).status == "ready"
    +    finally:
    +        conn.close()
    +
    +
    +def test_worker_block_rejects_foreign_task_id(worker_env):
    +    """A worker cannot block a task that isn't its own (#19534)."""
    +    from hermes_cli import kanban_db as kb
    +    conn = kb.connect()
    +    try:
    +        other = kb.create_task(conn, title="sibling")
    +        conn.execute("UPDATE tasks SET status='ready' WHERE id=?", (other,))
    +        conn.commit()
    +    finally:
    +        conn.close()
    +
    +    from tools import kanban_tools as kt
    +    out = kt._handle_block({"task_id": other, "reason": "evil"})
    +    d = json.loads(out)
    +    assert "refusing to mutate" in d.get("error", "")
    +
    +    conn = kb.connect()
    +    try:
    +        assert kb.get_task(conn, other).status == "ready"
    +    finally:
    +        conn.close()
    +
    +
    +def test_worker_heartbeat_rejects_foreign_task_id(worker_env):
    +    """A worker cannot heartbeat a task that isn't its own (#19534)."""
    +    from hermes_cli import kanban_db as kb
    +    conn = kb.connect()
    +    try:
    +        other = kb.create_task(conn, title="sibling")
    +        # Put sibling in running state so heartbeat would otherwise succeed.
    +        conn.execute("UPDATE tasks SET status='running' WHERE id=?", (other,))
    +        conn.commit()
    +    finally:
    +        conn.close()
    +
    +    from tools import kanban_tools as kt
    +    out = kt._handle_heartbeat({"task_id": other})
    +    d = json.loads(out)
    +    assert "refusing to mutate" in d.get("error", "")
    +
    +
    +def test_worker_complete_own_task_still_works(worker_env):
    +    """The ownership check doesn't break the normal own-task happy path."""
    +    from tools import kanban_tools as kt
    +    # Both implicit (no task_id arg) and explicit (matching env) must work.
    +    out = kt._handle_complete({"task_id": worker_env, "summary": "explicit own"})
    +    d = json.loads(out)
    +    assert d.get("ok") is True and d.get("task_id") == worker_env
    +
    +
    +def test_orchestrator_complete_any_task_allowed(monkeypatch, tmp_path):
    +    """Orchestrator profiles (no HERMES_KANBAN_TASK) can still complete
    +    any task via explicit task_id. The check only applies to workers."""
    +    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
    +    home = tmp_path / ".hermes"
    +    home.mkdir()
    +    monkeypatch.setenv("HERMES_HOME", str(home))
    +    from pathlib import Path as _P
    +    monkeypatch.setattr(_P, "home", lambda: tmp_path)
    +
    +    from hermes_cli import kanban_db as kb
    +    kb._INITIALIZED_PATHS.clear()
    +    kb.init_db()
    +    conn = kb.connect()
    +    try:
    +        tid = kb.create_task(conn, title="child to close out")
    +        conn.execute("UPDATE tasks SET status='ready' WHERE id=?", (tid,))
    +        conn.commit()
    +    finally:
    +        conn.close()
    +
    +    from tools import kanban_tools as kt
    +    out = kt._handle_complete({"task_id": tid, "summary": "orchestrator close"})
    +    d = json.loads(out)
    +    assert d.get("ok") is True and d.get("task_id") == tid

  • tools/kanban_tools.py

    @@ -79,6 +79,38 @@ def _default_task_id(arg: Optional[str]) -> Optional[str]:
         return env_tid or None
     
     
    +def _enforce_worker_task_ownership(tid: str) -> Optional[str]:
    +    """Reject worker-driven destructive calls on foreign task IDs.
    +
    +    A process spawned by the dispatcher has ``HERMES_KANBAN_TASK`` set
    +    to its own task id. Tools like ``kanban_complete`` / ``kanban_block``
    +    / ``kanban_heartbeat`` mutate run-lifecycle state, so a buggy or
    +    prompt-injected worker that passed an explicit ``task_id`` for some
    +    other task could corrupt sibling or cross-tenant runs (see #19534).
    +
    +    Orchestrator profiles (kanban toolset enabled but **no**
    +    ``HERMES_KANBAN_TASK`` in env) aren't subject to this check — their
    +    job is routing, and they sometimes legitimately close out child
    +    tasks or reopen blocked ones. Workers are narrowly scoped to their
    +    one task.
    +
    +    Returns ``None`` when the call is allowed, or a tool-error string
    +    when it must be rejected. Callers should ``return`` the error
    +    verbatim.
    +    """
    +    env_tid = os.environ.get("HERMES_KANBAN_TASK")
    +    if not env_tid:
    +        # Orchestrator or CLI context — no task-scope restriction.
    +        return None
    +    if tid != env_tid:
    +        return tool_error(
    +            f"worker is scoped to task {env_tid}; refusing to mutate "
    +            f"{tid}. Use kanban_comment to hand off information to other "
    +            f"tasks, or kanban_create to spawn follow-up work."
    +        )
    +    return None
    +
    +
     def _connect():
         """Import + connect lazily so the module imports cleanly in non-kanban
         contexts (e.g. test rigs that import every tool module)."""
    @@ -172,6 +204,9 @@ def _handle_complete(args: dict, **kw) -> str:
             return tool_error(
                 "task_id is required (or set HERMES_KANBAN_TASK in the env)"
             )
    +    ownership_err = _enforce_worker_task_ownership(tid)
    +    if ownership_err:
    +        return ownership_err
         summary = args.get("summary")
         metadata = args.get("metadata")
         result = args.get("result")
    @@ -210,6 +245,9 @@ def _handle_block(args: dict, **kw) -> str:
             return tool_error(
                 "task_id is required (or set HERMES_KANBAN_TASK in the env)"
             )
    +    ownership_err = _enforce_worker_task_ownership(tid)
    +    if ownership_err:
    +        return ownership_err
         reason = args.get("reason")
         if not reason or not str(reason).strip():
             return tool_error("reason is required — explain what input you need")
    @@ -238,6 +276,9 @@ def _handle_heartbeat(args: dict, **kw) -> str:
             return tool_error(
                 "task_id is required (or set HERMES_KANBAN_TASK in the env)"
             )
    +    ownership_err = _enforce_worker_task_ownership(tid)
    +    if ownership_err:
    +        return ownership_err
         note = args.get("note")
         try:
             kb, conn = _connect()