fix(tasks): apply the admin-only action policy in manage_tasks (#6358)

POST /api/tasks and the scheduler both refuse action tasks whose action
is in ADMIN_ONLY_TASK_ACTIONS (run_local, run_script, ssh_command,
cookbook_serve) for owners without admin task privileges: those actions
reach subprocess.run(shell=True) or SSH with no sandbox. The model-facing
manage_tasks tool applied neither check, so an agent could create or edit
such a task, resume it, or trigger it, and only the scheduler's
execution-time guard stood in the way (pausing the task after the fact).

Route the tool through the same src.task_action_policy helpers the HTTP
route uses: create checks the requested type/action, edit checks the task
as it would be after the edit (so an llm task cannot be flipped into a
shell action), and resume/run check the stored task. Admins, non-
privileged actions and llm/research tasks are unaffected; pause still
works for everyone since it only narrows.

Fixes #6021


Claude-Session: https://claude.ai/code/session_01UAB5sz5srMKR7rbeDFtdMH

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
Co-authored-by: Alexandre Teixeira <111787685+alteixeira20@users.noreply.github.com>

src/tools/system.py

@@ -325,6 +325,26 @@ def _task_structured_schedule(args, fallback_time=None):
             'cron_expression': f'{parsed.minute} {parsed.hour} * * {cron_days}'}
 
 
+def _admin_task_action_denied(owner: Optional[str], task_type: Optional[str], action_name: Optional[str]) -> Optional[Dict]:
+    """Apply the shared admin-only task-action policy to a tool call.
+
+    ``POST /api/tasks`` and the scheduler both refuse ``run_local``,
+    ``run_script``, ``ssh_command`` and ``cookbook_serve`` action tasks for
+    owners without admin task privileges (``src.task_action_policy``).
+    Those actions reach ``subprocess.run(shell=True)`` / SSH with no sandbox,
+    so the model-facing tool must apply the same policy at create/edit/
+    resume/run time instead of storing a task the scheduler will only pause
+    later. Returns the tool error dict to hand back, or None when allowed.
+    """
+    from src.task_action_policy import (
+        is_admin_only_task_action,
+        owner_has_admin_task_privileges,
+    )
+    if is_admin_only_task_action(task_type, action_name) and not owner_has_admin_task_privileges(owner):
+        return {"error": f"Action '{action_name}' requires admin privileges", "exit_code": 1}
+    return None
+
+
 async def do_manage_tasks(content: str, owner: Optional[str] = None) -> Dict:
     """Handle manage_tasks tool calls: CRUD on scheduled tasks."""
     import uuid as _uuid
@@ -462,6 +482,9 @@ async def do_manage_tasks(content: str, owner: Optional[str] = None) -> Dict:
                 return {"error": "Prompt is required for llm/research tasks", "exit_code": 1}
             if task_type == "action" and not args.get("action_name"):
                 return {"error": "action_name is required for action tasks", "exit_code": 1}
+            denied = _admin_task_action_denied(owner, task_type, args.get("action_name"))
+            if denied:
+                return denied
 
             # Compute next_run for schedule triggers
             next_run = None
@@ -525,6 +548,14 @@ async def do_manage_tasks(content: str, owner: Optional[str] = None) -> Dict:
             if owner and task.owner != owner:
                 return {"error": "Access denied", "exit_code": 1}
 
+            # Check the policy against the task as it would be after the
+            # edit, so an llm task can't be flipped into a shell action.
+            next_task_type = args["task_type"] if args.get("task_type") is not None else task.task_type
+            next_action = args["action_name"] if args.get("action_name") is not None else task.action
+            denied = _admin_task_action_denied(owner, next_task_type, next_action)
+            if denied:
+                return denied
+
             if 'weekdays' in args or 'day_of_month' in args:
                 clock = task.scheduled_time
                 if task.schedule == 'cron':
@@ -624,6 +655,9 @@ async def do_manage_tasks(content: str, owner: Optional[str] = None) -> Dict:
             if action == "pause":
                 task.status = "paused"
             else:
+                denied = _admin_task_action_denied(owner, task.task_type, task.action)
+                if denied:
+                    return denied
                 task.status = "active"
                 if (task.trigger_type or "schedule") == "schedule":
                     task.next_run = compute_next_run(
@@ -644,6 +678,9 @@ async def do_manage_tasks(content: str, owner: Optional[str] = None) -> Dict:
                 return error
             if owner and task.owner != owner:
                 return {"error": "Access denied", "exit_code": 1}
+            denied = _admin_task_action_denied(owner, task.task_type, task.action)
+            if denied:
+                return denied
 
             from src.event_bus import get_task_scheduler
             scheduler = get_task_scheduler()

tests/test_manage_tasks_admin_action_policy.py

@@ -0,0 +1,289 @@
+"""manage_tasks must apply the admin-only task-action policy.
+
+``POST /api/tasks`` (routes/task/task_routes.py) and the scheduler
+(src/task_scheduler.py) both refuse ``action`` tasks whose action is in
+``ADMIN_ONLY_TASK_ACTIONS`` (run_local, run_script, ssh_command,
+cookbook_serve) for owners without admin task privileges — those actions run
+``subprocess.run(shell=True)`` or SSH with no sandbox. The model-facing
+``do_manage_tasks`` applied neither check, so an agent could store such a task
+on create/edit and resume/run it; the scheduler would only pause it at
+execution time. Regression for issue #6021: the tool now enforces the same
+policy on create, edit, resume and run, and keeps allowing admins and the
+non-privileged actions.
+"""
+
+import json
+import tempfile
+
+import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from sqlalchemy.pool import NullPool
+
+from tests.helpers.import_state import clear_fake_database_modules
+
+clear_fake_database_modules()
+
+import core.database as cdb
+from core.database import ScheduledTask
+import src.task_action_policy as policy_mod
+import src.tools.system as system_mod
+from src.tools.system import do_manage_tasks
+
+_TMPDB = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
+_ENGINE = create_engine(
+    f"sqlite:///{_TMPDB.name}",
+    connect_args={"check_same_thread": False},
+    poolclass=NullPool,
+)
+cdb.Base.metadata.create_all(_ENGINE)
+_TS = sessionmaker(bind=_ENGINE, autoflush=False, autocommit=False)
+
+
+@pytest.fixture(autouse=True)
+def _bind_session_factory(monkeypatch):
+    # do_manage_tasks does `from core.database import SessionLocal` at call
+    # time. Bind it per test (not at import) so this file stays
+    # order-independent from sibling files that point it at their own DB.
+    monkeypatch.setattr(cdb, "SessionLocal", _TS)
+
+
+@pytest.fixture
+def privileges(monkeypatch):
+    """Drive owner_has_admin_task_privileges from a plain dict."""
+    admins = {"root"}
+    monkeypatch.setattr(
+        policy_mod, "owner_has_admin_task_privileges",
+        lambda owner: owner in admins,
+    )
+    return admins
+
+
+def _seed(task_id, owner, task_type="llm", action=None, status="active"):
+    db = _TS()
+    try:
+        db.add(ScheduledTask(
+            id=task_id, owner=owner, name=task_id, prompt="original",
+            task_type=task_type, action=action, trigger_type="webhook",
+            status=status, output_target="session",
+        ))
+        db.commit()
+    finally:
+        db.close()
+
+
+def _get(task_id):
+    db = _TS()
+    try:
+        return db.query(ScheduledTask).filter(ScheduledTask.id == task_id).first()
+    finally:
+        db.close()
+
+
+def _count(owner):
+    db = _TS()
+    try:
+        return db.query(ScheduledTask).filter(ScheduledTask.owner == owner).count()
+    finally:
+        db.close()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("action_name", sorted(policy_mod.ADMIN_ONLY_TASK_ACTIONS))
+async def test_create_admin_only_action_denied_for_non_admin(privileges, action_name):
+    before = _count("alice")
+    out = await do_manage_tasks(json.dumps({
+        "action": "create", "task_type": "action", "action_name": action_name,
+        "name": "pwn", "schedule": "daily", "scheduled_time": "09:00",
+    }), owner="alice")
+    assert out["exit_code"] == 1, out
+    assert "requires admin privileges" in out["error"]
+    assert _count("alice") == before
+
+
+@pytest.mark.asyncio
+async def test_create_admin_only_action_allowed_for_admin(privileges):
+    out = await do_manage_tasks(json.dumps({
+        "action": "create", "task_type": "action", "action_name": "run_local",
+        "name": "ok", "schedule": "daily", "scheduled_time": "09:00",
+    }), owner="root")
+    assert out["exit_code"] == 0, out
+    assert _get(out["task_id"]).action == "run_local"
+
+
+@pytest.mark.asyncio
+async def test_create_regular_action_allowed_for_non_admin(privileges):
+    out = await do_manage_tasks(json.dumps({
+        "action": "create", "task_type": "action", "action_name": "tidy_sessions",
+        "name": "ok", "schedule": "daily", "scheduled_time": "09:00",
+    }), owner="alice")
+    assert out["exit_code"] == 0, out
+
+
+@pytest.mark.asyncio
+async def test_create_llm_task_allowed_for_non_admin(privileges):
+    out = await do_manage_tasks(json.dumps({
+        "action": "create", "task_type": "llm", "prompt": "summarize",
+        "schedule": "daily", "scheduled_time": "09:00",
+    }), owner="alice")
+    assert out["exit_code"] == 0, out
+
+
+@pytest.mark.asyncio
+async def test_edit_cannot_turn_llm_task_into_shell_action(privileges):
+    _seed("alice-llm", "alice")
+    out = await do_manage_tasks(json.dumps({
+        "action": "edit", "task_id": "alice-llm",
+        "task_type": "action", "action_name": "ssh_command",
+    }), owner="alice")
+    assert out["exit_code"] == 1, out
+    assert "requires admin privileges" in out["error"]
+    task = _get("alice-llm")
+    assert task.task_type == "llm" and task.action is None
+
+
+@pytest.mark.asyncio
+async def test_edit_cannot_swap_action_name_to_admin_only(privileges):
+    _seed("alice-action", "alice", task_type="action", action="tidy_sessions")
+    out = await do_manage_tasks(json.dumps({
+        "action": "edit", "task_id": "alice-action", "action_name": "run_script",
+    }), owner="alice")
+    assert out["exit_code"] == 1, out
+    assert _get("alice-action").action == "tidy_sessions"
+
+
+@pytest.mark.asyncio
+async def test_edit_of_unrelated_field_on_admin_only_task_denied_for_non_admin(privileges):
+    # A task that slipped in before the policy (or via demotion of its owner)
+    # cannot be kept alive by a non-admin: the route rejects the same edit.
+    _seed("alice-legacy", "alice", task_type="action", action="run_local")
+    out = await do_manage_tasks(json.dumps({
+        "action": "edit", "task_id": "alice-legacy", "name": "renamed",
+    }), owner="alice")
+    assert out["exit_code"] == 1, out
+    assert _get("alice-legacy").name == "alice-legacy"
+
+
+@pytest.mark.asyncio
+async def test_resume_admin_only_task_denied_for_non_admin(privileges):
+    _seed("alice-paused", "alice", task_type="action", action="run_local", status="paused")
+    out = await do_manage_tasks(json.dumps({
+        "action": "resume", "task_id": "alice-paused",
+    }), owner="alice")
+    assert out["exit_code"] == 1, out
+    assert _get("alice-paused").status == "paused"
+
+
+@pytest.mark.asyncio
+async def test_pause_admin_only_task_still_allowed(privileges):
+    # Narrowing is always fine: a non-admin may pause a task they own.
+    _seed("alice-active", "alice", task_type="action", action="run_local")
+    out = await do_manage_tasks(json.dumps({
+        "action": "pause", "task_id": "alice-active",
+    }), owner="alice")
+    assert out["exit_code"] == 0, out
+    assert _get("alice-active").status == "paused"
+
+
+@pytest.mark.asyncio
+async def test_run_admin_only_task_denied_before_scheduler(privileges, monkeypatch):
+    _seed("alice-run", "alice", task_type="action", action="ssh_command")
+    calls = []
+
+    class _Sched:
+        async def run_task_now(self, task_id):
+            calls.append(task_id)
+            return True
+
+    import src.event_bus as event_bus
+    monkeypatch.setattr(event_bus, "get_task_scheduler", lambda: _Sched())
+    out = await do_manage_tasks(json.dumps({
+        "action": "run", "task_id": "alice-run",
+    }), owner="alice")
+    assert out["exit_code"] == 1, out
+    assert calls == []
+
+
+@pytest.mark.asyncio
+async def test_run_admin_only_task_allowed_for_admin(privileges, monkeypatch):
+    _seed("root-run", "root", task_type="action", action="ssh_command")
+    calls = []
+
+    class _Sched:
+        async def run_task_now(self, task_id):
+            calls.append(task_id)
+            return True
+
+    import src.event_bus as event_bus
+    monkeypatch.setattr(event_bus, "get_task_scheduler", lambda: _Sched())
+    out = await do_manage_tasks(json.dumps({
+        "action": "run", "task_id": "root-run",
+    }), owner="root")
+    assert out["exit_code"] == 0, out
+    assert calls == ["root-run"]
+
+
+def test_policy_helper_uses_shared_constants(privileges):
+    # The tool must consult the shared policy module, not a private copy of
+    # the action names, so a new admin-only action is covered everywhere.
+    assert system_mod._admin_task_action_denied("alice", "action", "run_local") is not None
+    assert system_mod._admin_task_action_denied("root", "action", "run_local") is None
+    assert system_mod._admin_task_action_denied("alice", "action", "tidy_sessions") is None
+    assert system_mod._admin_task_action_denied("alice", "llm", "run_local") is None
+
+
+@pytest.mark.asyncio
+async def test_run_admin_only_task_by_name_denied_for_non_admin(privileges, monkeypatch):
+    _seed("alice-run-name", "alice", task_type="action", action="ssh_command")
+    calls = []
+
+    class _Sched:
+        async def run_task_now(self, task_id):
+            calls.append(task_id)
+            return True
+
+    import src.event_bus as event_bus
+    monkeypatch.setattr(event_bus, "get_task_scheduler", lambda: _Sched())
+    out = await do_manage_tasks(json.dumps({
+        "action": "run", "name": "alice-run-name",
+    }), owner="alice")
+    assert out["exit_code"] == 1, out
+    assert "requires admin privileges" in out["error"]
+    assert calls == []
+
+
+@pytest.mark.asyncio
+async def test_create_admin_only_action_seals_authority_for_admin(privileges):
+    out = await do_manage_tasks(json.dumps({
+        "action": "create", "task_type": "action", "action_name": "run_local",
+        "name": "sealed-task", "schedule": "daily", "scheduled_time": "09:00",
+    }), owner="root")
+    assert out["exit_code"] == 0, out
+    task = _get(out["task_id"])
+    assert task.action == "run_local"
+    assert task.request_authority_json is not None
+
+
+def test_real_owner_has_admin_task_privileges_branches(monkeypatch):
+    """Verify owner_has_admin_task_privileges without mocking."""
+    # Auth disabled branch
+    monkeypatch.setenv("AUTH_ENABLED", "false")
+    assert policy_mod.owner_has_admin_task_privileges("alice") is True
+    assert policy_mod.owner_has_admin_task_privileges(None) is True
+
+    # Auth enabled branch with configured AuthManager
+    monkeypatch.delenv("AUTH_ENABLED", raising=False)
+
+    class FakeAuth:
+        is_configured = True
+
+        def is_admin(self, u):
+            return u == "admin"
+
+    import core.auth as c_auth
+    monkeypatch.setattr(c_auth, "AuthManager", lambda: FakeAuth())
+
+    assert policy_mod.owner_has_admin_task_privileges("admin") is True
+    assert policy_mod.owner_has_admin_task_privileges("alice") is False
+    assert policy_mod.owner_has_admin_task_privileges("internal-tool") is True
+    assert policy_mod.owner_has_admin_task_privileges(None) is False