mirror of
https://github.com/andrewyng/openworker.git
synced 2026-11-04 13:27:11 +00:00
Darryl Pentz
8ea2dc32f4
"Allow every time" on a run's approval card mints a standing rule on the task, but
only when the call is external-risk AND names a target. A routine whose work is web
search therefore asks the same question on every single run: the session-scoped
"always" dies with the run, and each scheduled run is a fresh session.
The refusal is upside down against the engine's own strictness table. Egress ranks
1, external ranks 2 — so a routine can be trusted forever to send a message off the
machine, but never to read a web page. The target binding is what makes an external
grant safe; a tool that takes no destination simply fell outside the shape and got
downgraded to a one-off, silently, while the API reported success.
Egress calls with no bindable target now mint a name-only rule on the task —
`ScheduledTask.add_tool_rule`, the same list, read by _seed_task_permissions on
every run and revocable through the existing PATCH …/{id} {"revoke": entry}.
Write-local and exec are unchanged (shell asks forever), and external stays
target-bound: "message this chat" is a grant, "send anything anywhere" is not.
Found running a daily monitoring routine on a headless box: it asked permission to
search the web every morning for four days, and "Always" could not stop it.
81 lines
3.0 KiB
Python
81 lines
3.0 KiB
Python
"""A routine can be trusted with a tool that has no target to bind.
|
|
|
|
"Always allow" on a run's approval card mints a standing rule on the task. That only
|
|
worked for external-risk calls naming a target, so a routine whose work is web search
|
|
asked the same question on every run: the session-scoped "always" dies with the run,
|
|
and each scheduled run is a fresh session. Egress ranks *below* external in the
|
|
engine's own strictness table, so the rule was being refused for the safer class.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from coworker.automation.models import ScheduledTask, Schedule
|
|
from coworker.permissions import standing_tool_candidate
|
|
from coworker.providers import ModelCapabilities, ProviderClient
|
|
from coworker.server.manager import SessionManager
|
|
|
|
|
|
class NoTurnsProvider(ProviderClient):
|
|
def complete(self, *, model, messages, tools=None, **settings):
|
|
raise AssertionError("no model turns expected")
|
|
|
|
def capabilities(self, model):
|
|
return ModelCapabilities()
|
|
|
|
|
|
class _Request:
|
|
def __init__(self, tool_name, arguments=None):
|
|
self.tool_name = tool_name
|
|
self.arguments = arguments or {}
|
|
self.metadata = None
|
|
|
|
|
|
def test_only_egress_earns_a_target_less_grant():
|
|
assert standing_tool_candidate("web_search") is True
|
|
assert standing_tool_candidate("web_fetch") is True
|
|
# Exec and local writes ask forever, as they do for target-bound rules.
|
|
assert standing_tool_candidate("run_shell") is False
|
|
assert standing_tool_candidate("write_file") is False
|
|
|
|
|
|
def _task_manager(tmp_path):
|
|
manager = SessionManager(data_dir=tmp_path / "data", provider=NoTurnsProvider())
|
|
task = ScheduledTask(
|
|
title="Daily watch",
|
|
instructions="sweep",
|
|
agent="watcher",
|
|
workspace=str(tmp_path / "work"),
|
|
schedule=Schedule(kind="cron", cron="0 7 * * *"),
|
|
)
|
|
manager.task_store.save(task)
|
|
run_session = task.task_session_id
|
|
return manager, task, run_session
|
|
|
|
|
|
def test_web_search_grant_persists_on_the_task(tmp_path):
|
|
manager, task, _ = _task_manager(tmp_path)
|
|
from coworker.automation.models import TaskRun
|
|
|
|
run = TaskRun(task_id=task.id)
|
|
manager.task_store.add_run(run)
|
|
|
|
assert manager.mint_task_rule(run.session_id, "web_search", {"query": "anything"})
|
|
|
|
fresh = manager.task_store.get(task.id)
|
|
assert "web_search" in fresh.name_allowed_tools()
|
|
# A name-only entry is revocable like any other, and reads back on the API shape.
|
|
entry = [r for r in fresh.public()["always_allowed"] if r["tool"] == "web_search"][0]
|
|
assert entry["target"] is None
|
|
assert fresh.revoke_rule(entry["entry"]) is True
|
|
|
|
|
|
def test_shell_still_asks_every_time(tmp_path):
|
|
manager, task, _ = _task_manager(tmp_path)
|
|
from coworker.automation.models import TaskRun
|
|
|
|
run = TaskRun(task_id=task.id)
|
|
manager.task_store.add_run(run)
|
|
|
|
assert manager.mint_task_rule(run.session_id, "run_shell", {"command": "ls"}) is False
|
|
assert manager.task_store.get(task.id).always_allowed_tools == []
|