From a8d215a390f30b20235d0c69d7ed63cf35bcb72e Mon Sep 17 00:00:00 2001 From: Ashvin <76151462+ashvinctrl@users.noreply.github.com> Date: Sat, 11 Jul 2026 06:15:14 +0530 Subject: [PATCH] fix(tasks): scope manage_tasks mutations to an exact task owner (#5264) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The edit/delete/pause/run actions of do_manage_tasks gated ownership with `if owner and task.owner and task.owner != owner`. The middle term made the check a no-op whenever task.owner was null/empty — the state a scheduled task sits in when it was created in no-login mode (or via the localhost middleware bypass) before the periodic legacy-owner sweep reassigns it to the admin user. Any authenticated user's agent could then edit, delete, pause, or run another tenant's owner-less task; edit+run lets an attacker rewrite the task prompt and execute it in the scheduler's agent context. The sibling `list` action already scopes with an exact `owner == owner` filter, so the mutators were strictly more permissive than the reader. Drop the middle term so the guard fails closed on owner-less rows for authenticated callers, matching `list` and the calendar/notes/gallery/session null-owner gates. Auth disabled (owner falsy) and same-owner access are unchanged. --- src/tools/system.py | 12 ++- tests/test_manage_tasks_owner_scope.py | 139 +++++++++++++++++++++++++ 2 files changed, 147 insertions(+), 4 deletions(-) create mode 100644 tests/test_manage_tasks_owner_scope.py diff --git a/src/tools/system.py b/src/tools/system.py index 3fedac6c3..a901992c2 100644 --- a/src/tools/system.py +++ b/src/tools/system.py @@ -356,7 +356,11 @@ async def do_manage_tasks(content: str, owner: Optional[str] = None) -> Dict: task = db.query(ScheduledTask).filter(ScheduledTask.id == task_id).first() if not task: return {"error": f"Task {task_id} not found", "exit_code": 1} - if owner and task.owner and task.owner != owner: + # Strict ownership: the old `task.owner and task.owner != owner` + # skipped the check on an owner-less task (created in no-login mode + # or before the legacy-owner sweep), letting any authenticated user + # reach it. `list` already scopes to an exact owner match. + if owner and task.owner != owner: return {"error": "Access denied", "exit_code": 1} changed = [] @@ -402,7 +406,7 @@ async def do_manage_tasks(content: str, owner: Optional[str] = None) -> Dict: task = db.query(ScheduledTask).filter(ScheduledTask.id == task_id).first() if not task: return {"error": f"Task {task_id} not found", "exit_code": 1} - if owner and task.owner and task.owner != owner: + if owner and task.owner != owner: return {"error": "Access denied", "exit_code": 1} name = task.name db.delete(task) @@ -416,7 +420,7 @@ async def do_manage_tasks(content: str, owner: Optional[str] = None) -> Dict: task = db.query(ScheduledTask).filter(ScheduledTask.id == task_id).first() if not task: return {"error": f"Task {task_id} not found", "exit_code": 1} - if owner and task.owner and task.owner != owner: + if owner and task.owner != owner: return {"error": "Access denied", "exit_code": 1} if action == "pause": @@ -437,7 +441,7 @@ async def do_manage_tasks(content: str, owner: Optional[str] = None) -> Dict: task = db.query(ScheduledTask).filter(ScheduledTask.id == task_id).first() if not task: return {"error": f"Task {task_id} not found", "exit_code": 1} - if owner and task.owner and task.owner != owner: + if owner and task.owner != owner: return {"error": "Access denied", "exit_code": 1} from src.event_bus import get_task_scheduler diff --git a/tests/test_manage_tasks_owner_scope.py b/tests/test_manage_tasks_owner_scope.py new file mode 100644 index 000000000..334698ef0 --- /dev/null +++ b/tests/test_manage_tasks_owner_scope.py @@ -0,0 +1,139 @@ +"""manage_tasks mutations must fail closed on owner-less / cross-owner tasks. + +The edit/delete/pause/run actions of ``do_manage_tasks`` previously gated with +``if owner and task.owner and task.owner != owner``. The middle term made the +check a no-op whenever the task had no owner — the state a scheduled task is in +when it was created in no-login mode (or via the localhost middleware bypass) +before the periodic legacy-owner sweep reassigns it to the admin user. So any +authenticated user's agent could edit, delete, pause, or *run* another tenant's +owner-less task. The sibling ``list`` action already scopes with an exact +``ScheduledTask.owner == owner`` filter, so the mutators were strictly more +permissive than the reader. +""" + +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 +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) +# do_manage_tasks does `from core.database import SessionLocal` at call time, +# so patching the module attribute is enough to point it at the temp DB. +cdb.SessionLocal = _TS + + +def _seed(task_id, owner): + db = _TS() + try: + db.add(ScheduledTask( + id=task_id, owner=owner, name=task_id, prompt="original", + task_type="llm", trigger_type="webhook", status="active", + 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() + + +@pytest.mark.asyncio +async def test_edit_denied_on_ownerless_task_for_authenticated_user(): + _seed("ownerless-edit", None) + out = await do_manage_tasks( + json.dumps({"action": "edit", "task_id": "ownerless-edit", "prompt": "pwned"}), + owner="alice", + ) + assert out["exit_code"] == 1 and out["error"] == "Access denied" + assert _get("ownerless-edit").prompt == "original" + + +@pytest.mark.asyncio +async def test_delete_denied_on_ownerless_task_for_authenticated_user(): + _seed("ownerless-del", None) + out = await do_manage_tasks( + json.dumps({"action": "delete", "task_id": "ownerless-del"}), + owner="alice", + ) + assert out["exit_code"] == 1 and out["error"] == "Access denied" + assert _get("ownerless-del") is not None + + +@pytest.mark.asyncio +async def test_pause_denied_on_ownerless_task_for_authenticated_user(): + _seed("ownerless-pause", None) + out = await do_manage_tasks( + json.dumps({"action": "pause", "task_id": "ownerless-pause"}), + owner="alice", + ) + assert out["exit_code"] == 1 and out["error"] == "Access denied" + assert _get("ownerless-pause").status == "active" + + +@pytest.mark.asyncio +async def test_run_denied_on_ownerless_task_for_authenticated_user(): + _seed("ownerless-run", None) + out = await do_manage_tasks( + json.dumps({"action": "run", "task_id": "ownerless-run"}), + owner="alice", + ) + assert out["exit_code"] == 1 and out["error"] == "Access denied" + + +@pytest.mark.asyncio +async def test_edit_denied_on_other_owners_task(): + _seed("bob-task", "bob") + out = await do_manage_tasks( + json.dumps({"action": "edit", "task_id": "bob-task", "prompt": "pwned"}), + owner="alice", + ) + assert out["exit_code"] == 1 and out["error"] == "Access denied" + assert _get("bob-task").prompt == "original" + + +@pytest.mark.asyncio +async def test_edit_allowed_for_matching_owner(): + _seed("alice-task", "alice") + out = await do_manage_tasks( + json.dumps({"action": "edit", "task_id": "alice-task", "prompt": "updated"}), + owner="alice", + ) + assert out["exit_code"] == 0 + assert _get("alice-task").prompt == "updated" + + +@pytest.mark.asyncio +async def test_edit_allowed_in_no_login_mode(): + # owner is None when auth is disabled — single-user mode keeps full access + # to shared (owner-less) tasks, exactly as `list` returns them unfiltered. + _seed("shared-task", None) + out = await do_manage_tasks( + json.dumps({"action": "edit", "task_id": "shared-task", "prompt": "updated"}), + owner=None, + ) + assert out["exit_code"] == 0 + assert _get("shared-task").prompt == "updated"