diff --git a/src/tools/system.py b/src/tools/system.py index a5e9d340b..9b0e6d1bb 100644 --- a/src/tools/system.py +++ b/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() diff --git a/tests/test_manage_tasks_admin_action_policy.py b/tests/test_manage_tasks_admin_action_policy.py new file mode 100644 index 000000000..b30446d62 --- /dev/null +++ b/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