Nick
b7345f000f
CRITICAL: chat-originated tool dispatch (Python local_tool_runtime.py) had ZERO mode awareness — a chat write_file in CHAT mode executed ungated (the Go serve surface gated but the chat path never touched it). Closed: mode authorization at request intake in local_tool_ runtime, authorize seam in tool_protocol dispatch, approve() resume door gated, hatch-exposure filtered. Typed ModeRefusal feeds the model as tool-error data; failure breaker trips on repeat. T1: ONE mode matrix artifact (schemas/mode/v1/mode_matrix.json) read by BOTH Go (embed, byte-identical, 16-case battery) and Python (mode_gate.py, per-dispatch read 0.011ms, RUNE_MODE env wins). T2: model only SEES mode-allowed tools at capability selection. T3: per-dispatch mode re-check (mid-loop switch pinned). T4: tool events persist as channel-scoped role=tool conversation records — results land in the chat lane through the bridge poll. Main-session verify: mode dispatch gate + tool authority suites green own runs. 31/31 + 3438/0 new-mode-gating failures.
284 lines
11 KiB
Python
284 lines
11 KiB
Python
"""Authority-boundary tests for local tools.
|
|
|
|
These tests intentionally exercise the public prepare/approve/reject API rather
|
|
than UI confirmation code. A model may prepare a mutation, but only an
|
|
approval may perform it.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
|
|
import pytest
|
|
import json
|
|
from dataclasses import replace
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / "src"))
|
|
|
|
from rune import tool_executor # noqa: E402
|
|
from rune.local_tool_runtime import LocalToolRuntime # noqa: E402
|
|
from rune.tool_protocol import ToolRequest, validate_tool_request # noqa: E402
|
|
|
|
|
|
# wave-tools-live T3: the dispatch gate is always ON (absent mode state =
|
|
# CHAT). This battery pins the approval machinery — TASK-lane behavior.
|
|
@pytest.fixture(autouse=True)
|
|
def _pin_task_mode(monkeypatch):
|
|
monkeypatch.setenv("RUNE_MODE", "TASK")
|
|
|
|
|
|
class ToolAuthorityTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.tempdir = tempfile.TemporaryDirectory()
|
|
self.root = Path(self.tempdir.name) / "allowed"
|
|
self.root.mkdir()
|
|
self.undo = Path(self.tempdir.name) / "undo"
|
|
self.patches = [
|
|
patch.object(tool_executor, "ALLOWED_READ_DIRS", [str(self.root)]),
|
|
patch.object(tool_executor, "ALLOWED_WRITE_DIRS", [str(self.root)]),
|
|
patch.object(tool_executor, "BLOCKED_DIRS", []),
|
|
patch.object(tool_executor, "UNDO_DIR", str(self.undo)),
|
|
]
|
|
for item in self.patches:
|
|
item.start()
|
|
|
|
def tearDown(self) -> None:
|
|
for item in reversed(self.patches):
|
|
item.stop()
|
|
self.tempdir.cleanup()
|
|
|
|
def prepare(self, name: str, arguments: dict):
|
|
result = tool_executor.prepare_tool(name, arguments)
|
|
self.assertTrue(result["success"], result)
|
|
self.assertTrue(result["needs_confirm"], result)
|
|
return result["pending"]
|
|
|
|
def test_explicit_authority_sets_fail_closed_for_new_tools(self):
|
|
self.assertIn("read_file", tool_executor.READ_ONLY_TOOLS)
|
|
for name in {
|
|
"write_file",
|
|
"edit_file",
|
|
"format_code",
|
|
"git_commit",
|
|
"run_command",
|
|
}:
|
|
self.assertIn(name, tool_executor.MUTATING_TOOLS)
|
|
self.assertNotIn(name, tool_executor.READ_ONLY_TOOLS)
|
|
|
|
prepared = tool_executor.prepare_tool("future_tool", {"value": 1})
|
|
|
|
self.assertTrue(prepared["needs_confirm"])
|
|
self.assertFalse(tool_executor.approve_tool(prepared["pending"])["success"])
|
|
|
|
def test_read_tools_execute_during_prepare(self):
|
|
target = self.root / "note.txt"
|
|
target.write_text("hello", encoding="utf-8")
|
|
|
|
result = tool_executor.prepare_tool("read_file", {"path": str(target)})
|
|
|
|
self.assertTrue(result["success"])
|
|
self.assertNotIn("needs_confirm", result)
|
|
self.assertIn("hello", result["result"])
|
|
|
|
def test_each_mutator_preserves_bytes_until_approval(self):
|
|
target = self.root / "note.py"
|
|
original = "value=1\n"
|
|
target.write_text(original, encoding="utf-8")
|
|
|
|
cases = {
|
|
"write_file": {"path": str(target), "content": "value = 2\n"},
|
|
"edit_file": {"path": str(target), "search": "1", "replace": "2"},
|
|
"format_code": {"path": str(target), "language": "python"},
|
|
"git_commit": {"message": "model commit"},
|
|
"run_command": {"argv": ["python", "-c", "print('ok')"]},
|
|
}
|
|
|
|
with patch("subprocess.run") as run:
|
|
for name, arguments in cases.items():
|
|
with self.subTest(name=name):
|
|
result = tool_executor.prepare_tool(name, arguments)
|
|
self.assertTrue(result["needs_confirm"], result)
|
|
self.assertEqual(target.read_bytes(), original.encode())
|
|
run.assert_not_called()
|
|
|
|
def test_write_and_edit_approve_exactly_once_and_keep_backups(self):
|
|
target = self.root / "note.txt"
|
|
target.write_text("old", encoding="utf-8")
|
|
|
|
edit = self.prepare("edit_file", {"path": str(target), "search": "old", "replace": "mid"})
|
|
self.assertTrue(tool_executor.approve_tool(edit)["success"])
|
|
self.assertEqual(target.read_text(encoding="utf-8"), "mid")
|
|
self.assertFalse(tool_executor.approve_tool(edit)["success"])
|
|
|
|
write = self.prepare("write_file", {"path": str(target), "content": "new"})
|
|
self.assertTrue(tool_executor.approve_tool(write)["success"])
|
|
self.assertEqual(target.read_text(encoding="utf-8"), "new")
|
|
self.assertTrue(list(self.undo.glob("*.bak")))
|
|
|
|
def test_reject_and_expiry_preserve_bytes(self):
|
|
target = self.root / "note.txt"
|
|
target.write_text("old", encoding="utf-8")
|
|
rejected = self.prepare("write_file", {"path": str(target), "content": "rejected"})
|
|
expired = self.prepare("edit_file", {"path": str(target), "search": "old", "replace": "late"})
|
|
|
|
self.assertFalse(tool_executor.reject_tool(rejected)["success"])
|
|
self.assertFalse(tool_executor.approve_tool(replace(expired, expires_at=expired.created_at - 1))["success"])
|
|
self.assertEqual(target.read_bytes(), b"old")
|
|
|
|
def test_path_traversal_symlink_and_sibling_prefix_are_rejected(self):
|
|
outside = Path(self.tempdir.name) / "outside.txt"
|
|
outside.write_text("secret", encoding="utf-8")
|
|
sibling = Path(f"{self.root}-evil")
|
|
sibling.mkdir()
|
|
link = self.root / "link"
|
|
link.symlink_to(outside)
|
|
|
|
paths = [
|
|
self.root / ".." / "outside.txt",
|
|
sibling / "escape.txt",
|
|
link,
|
|
]
|
|
for path in paths:
|
|
with self.subTest(path=path):
|
|
result = tool_executor.prepare_tool("write_file", {"path": str(path), "content": "owned"})
|
|
self.assertFalse(result["success"], result)
|
|
self.assertEqual(outside.read_bytes(), b"secret")
|
|
|
|
def test_approval_revalidates_path_and_file_digest_against_toctou(self):
|
|
target = self.root / "note.txt"
|
|
target.write_text("old", encoding="utf-8")
|
|
pending = self.prepare("edit_file", {"path": str(target), "search": "old", "replace": "new"})
|
|
target.write_text("changed after preview", encoding="utf-8")
|
|
|
|
result = tool_executor.approve_tool(pending)
|
|
|
|
self.assertFalse(result["success"])
|
|
self.assertEqual(target.read_bytes(), b"changed after preview")
|
|
|
|
def test_approval_rejects_symlink_swap_after_preview(self):
|
|
target = self.root / "note.txt"
|
|
outside = Path(self.tempdir.name) / "outside.txt"
|
|
target.write_text("inside", encoding="utf-8")
|
|
outside.write_text("outside", encoding="utf-8")
|
|
pending = self.prepare("write_file", {"path": str(target), "content": "replacement"})
|
|
target.unlink()
|
|
target.symlink_to(outside)
|
|
|
|
result = tool_executor.approve_tool(pending)
|
|
|
|
self.assertFalse(result["success"])
|
|
self.assertEqual(outside.read_bytes(), b"outside")
|
|
|
|
def test_approval_rejects_tampered_operation_digest(self):
|
|
target = self.root / "note.txt"
|
|
target.write_text("old", encoding="utf-8")
|
|
pending = self.prepare("write_file", {"path": str(target), "content": "approved bytes"})
|
|
pending.arguments["content"] = "tampered bytes"
|
|
|
|
result = tool_executor.approve_tool(pending)
|
|
|
|
self.assertFalse(result["success"])
|
|
self.assertEqual(target.read_bytes(), b"old")
|
|
|
|
def test_argv_command_uses_shell_false_only_after_approval(self):
|
|
validate_tool_request(
|
|
ToolRequest(
|
|
"command-1",
|
|
"run_command",
|
|
{"argv": ["printf", "safe"]},
|
|
"ollama-native",
|
|
)
|
|
)
|
|
pending = self.prepare("run_command", {"argv": ["printf", "safe"]})
|
|
completed = subprocess.CompletedProcess(["printf", "safe"], 0, "safe", "")
|
|
|
|
with patch("subprocess.run", return_value=completed) as run:
|
|
result = tool_executor.approve_tool(pending)
|
|
|
|
self.assertTrue(result["success"])
|
|
run.assert_called_once()
|
|
self.assertIs(run.call_args.kwargs["shell"], False)
|
|
self.assertEqual(run.call_args.args[0], ["printf", "safe"])
|
|
|
|
def test_shell_syntax_requires_explicit_approved_shell_path(self):
|
|
target = self.root / "command-output"
|
|
ordinary = self.prepare("run_command", {"command": f"printf owned > {target}"})
|
|
|
|
with patch("subprocess.run") as run:
|
|
denied = tool_executor.approve_tool(ordinary)
|
|
run.assert_not_called()
|
|
self.assertFalse(denied["success"])
|
|
self.assertFalse(target.exists())
|
|
|
|
explicit = self.prepare(
|
|
"run_command",
|
|
{"command": f"printf owned > {target}", "shell": True},
|
|
)
|
|
completed = subprocess.CompletedProcess("command", 0, "", "")
|
|
with patch("subprocess.run", return_value=completed) as run:
|
|
approved = tool_executor.approve_tool(explicit)
|
|
self.assertTrue(approved["success"])
|
|
self.assertIs(run.call_args.kwargs["shell"], True)
|
|
|
|
def test_local_runtime_prepares_and_approves_through_authority_api(self):
|
|
target = self.root / "runtime.txt"
|
|
target.write_text("old", encoding="utf-8")
|
|
|
|
class Transport:
|
|
def __init__(self):
|
|
self.responses = [
|
|
{
|
|
"role": "assistant",
|
|
"content": "",
|
|
"tool_calls": [
|
|
{
|
|
"id": "write-1",
|
|
"function": {
|
|
"name": "write_file",
|
|
"arguments": {
|
|
"path": str(target),
|
|
"content": "new",
|
|
},
|
|
},
|
|
}
|
|
],
|
|
},
|
|
{"role": "assistant", "content": "done"},
|
|
]
|
|
|
|
def supports_tools(self, model):
|
|
return True
|
|
|
|
def supports_thinking(self, model):
|
|
return False
|
|
|
|
def chat(self, **kwargs):
|
|
return self.responses.pop(0)
|
|
|
|
runtime = LocalToolRuntime(transport=Transport(), executor=tool_executor.execute_tool, model="local")
|
|
|
|
pending = runtime.run([{"role": "user", "content": "change it"}])
|
|
|
|
self.assertEqual(target.read_bytes(), b"old")
|
|
json.dumps(pending.pending_state)
|
|
self.assertIn("prepared_tool", pending.pending_state)
|
|
|
|
completed = runtime.approve(pending.pending_state)
|
|
|
|
self.assertTrue(completed.success)
|
|
self.assertEqual(target.read_bytes(), b"new")
|
|
duplicate = runtime.approve(pending.pending_state)
|
|
self.assertFalse(duplicate.success)
|
|
self.assertEqual(target.read_bytes(), b"new")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|