From ab8addd78f566137b3cd8a502af600332f63239c Mon Sep 17 00:00:00 2001 From: tyxben <15245012960@163.com> Date: Sat, 13 Jun 2026 20:21:35 +0800 Subject: [PATCH] fix(tool_guard): gate unknown/external WRITE tools by side-effect (close MCP bypass) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit gate() short-circuited any tool name outside a hardcoded 6-set to AUTO, so an MCP write tool (e.g. gmail.send_email) reaching the approval gate was auto-approved — the unknown-WRITE / tool-poisoning bypass (CVE-2025-54136 class). This is the keystone for MCP integration. confirmation_callback now passes spec.side_effect + requires_confirmation into gate(); gate() decides by side-effect first: - name-keyed native tools keep their curated policy (shell danger-matched; the always-confirm writers gate every call) - requires_confirmation=True always gates (native or not) - side_effect=write AND not a native tool (external/MCP) gates by default - native low-risk writes (reminders/todos/notes/voice) stay AUTO via a set_native_tools() snapshot taken at startup before any connect_mcp() — fixes the over-gating regression an adversarial review caught (all 8 in-repo writes would have modal-spammed in CONFIRM mode, with no allowlist escape) - malformed spec fails closed (treated as write) Adjacent bugs the review surfaced (latent on main): - _ALWAYS_CONFIRM had a dead "create_session" entry; the real tool is create_claude_session, so iTerm session-creation was never actually gated. Fixed the name + the _primary_text extractor. - enroll_face declared neither side_effect nor requires_confirmation, so it never reached the gate despite docs/CLAUDE.md claiming it is gated. Added side_effect="write". Verified against the real Arcana 1.0 contract: gateway._confirm_execution only invokes the callback for WRITE/requires_confirmation tools; MCP classifies side-effect via a name/desc keyword heuristic (default READ) — residual noted in CLAUDE.md (a write tool whose name dodges the keywords classifies READ and never reaches the gate; Arcana-level, fix later). 528 pytest green (+15), incl. an integration test that dispatches a WRITE tool through the real gateway to pin Arcana's routing contract. Reviewed by a 3-lens adversarial workflow (security-bypass / regression / correctness); the over-gating regression and adjacent bugs are fixed here. --- CLAUDE.md | 9 +- adapters/telegram_bot.py | 11 ++ server.py | 16 ++ tests/test_tool_guard.py | 252 +++++++++++++++++++++++++++ tests/test_tool_guard_integration.py | 50 ++++++ tool_guard.py | 161 ++++++++++++++--- tools/vision.py | 4 + 7 files changed, 476 insertions(+), 27 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 745b771..c25c175 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -139,7 +139,14 @@ Same shape as the soul review gate, but it gates the *action* path (the agent ca - `log` — non-dangerous calls pass through; dangerous calls (matched against `tool_guard.DANGEROUS_PATTERNS` — 38 curated regexes covering rm/dd/sudo/pipe-to-shell/credential dirs/Roboot at-rest paths/etc.) are still allowed but the call lands in `.tool_audit/--LOGGED.json`. - `confirm` — dangerous calls broadcast `{"type":"tool_approval","req_id":...,"tool":...,"args_summary":...,"danger_reason":...,"origin":...,"issued_at":...,"timeout_s":30}` to every registered surface (local console + relay mobile + Telegram). The reply `{"type":"tool_approval_decision","req_id":...,"approved":bool}` resolves the gate's pending future. No reply within `timeout_s` → REJECTED. Args summary > 2 KB → REJECTED unconditionally. -Hooked into Arcana via `runtime._tool_gateway.confirmation_callback = tool_guard.confirmation_callback` — each gated tool just declares `requires_confirmation=True` on its `@arcana.tool(...)` decorator (or `side_effect="write"`). Currently gated: `shell` (danger-pattern matched on the command), the iTerm write tools `send_to_session`/`create_session`, `enroll_face`, and the filesystem writers `write_file`/`edit_file` (always-confirm, keyed on the target path, allowlistable by path). The danger detector also catches locally-decoded payloads piped to a shell/interpreter (`base64 -d | sh`, `xxd -r | sh`, `cat x | sh`), not just download-anchored `curl | bash`. The callback fails *closed* on any internal exception (returns False) so a crashing gate rejects the call rather than waving it through. +Hooked into Arcana via `runtime._tool_gateway.confirmation_callback = tool_guard.confirmation_callback` — each gated tool just declares `requires_confirmation=True` on its `@arcana.tool(...)` decorator (or `side_effect="write"`). Gating is **side-effect-first, not a fixed name list** (`gate()` reads `spec.side_effect`/`requires_confirmation`, passed through by `confirmation_callback`): + +- **Name-keyed native tools** (curated policy): `shell` (danger-pattern matched on the command — only dangerous commands gate), and the always-confirm writers `send_to_session`, `create_claude_session`, `enroll_face`, `write_file`, `edit_file` (every call gated, keyed on the target path, allowlistable by path). These names are *reserved for native tools*. +- **Roboot's other native writes** (`schedule_reminder`/`add_todo`/`update_self`/`switch_tts_voice`/…) stay **AUTO** — they're registered via `tool_guard.set_native_tools(...)` at startup (snapshotted from the runtime's tool registry *before* any `connect_mcp()`) and are exempt from side-effect-first gating, preserving pre-MCP behavior (no modal-spam on benign reminders/todos). +- **Unknown / external WRITE tools** — anything Arcana flags `side_effect=WRITE` that is *not* a native tool, i.e. an **MCP write tool** like `gmail.send_email` — are **gated by default**. Without this an undeclared write would short-circuit to AUTO and bypass approval (the tool-name-allowlist hole; the MCP tool-poisoning / unknown-WRITE-bypass class, CVE-2025-54136). They have no `_primary_text` extractor so they can't be prefix-allowlisted today. +- A tool that explicitly declares `requires_confirmation=True` is **always gated**, native or not. + +Residual: Arcana's MCP layer classifies a tool's side-effect by a name/description keyword heuristic (`_infer_side_effect`, default READ), so a write tool whose name dodges the keywords classifies READ and never reaches the gate — an Arcana-level gap (fix later by registering MCP servers with explicit side-effects or the HEAD guardrail API). The danger detector also catches locally-decoded payloads piped to a shell/interpreter (`base64 -d | sh`, `xxd -r | sh`, `cat x | sh`), not just download-anchored `curl | bash`. The callback fails *closed* on any internal exception or malformed spec (returns False / treats unreadable spec as write) so a crashing gate rejects the call rather than waving it through. The danger detector applies NFKC + ANSI strip + null-byte normalization before matching, with a 16 KB hard cap on detector input (ReDoS guard). Allowlist at `~/.roboot/tool_allowlist.json` (per-machine, gitignored) does prefix matching with token boundaries; metachar-containing entries (`;`, `&`, backtick, `$(`, `||`, etc.) are silently rejected at lookup time so a user can't write `prefix: "ls; rm -rf"` and feel safe. Allowlist CANNOT override danger detection — a dangerous shell command goes to modal regardless. diff --git a/adapters/telegram_bot.py b/adapters/telegram_bot.py index 2f445a8..95b235a 100644 --- a/adapters/telegram_bot.py +++ b/adapters/telegram_bot.py @@ -169,6 +169,17 @@ def _get_runtime() -> arcana.Runtime: _runtime._tool_gateway.confirmation_callback = ( tool_guard.confirmation_callback ) + # Snapshot native tool names (see server.py for rationale): only + # UNKNOWN/external writes gate by default; native low-risk writes + # stay AUTO. Run before any connect_mcp(). + try: + tool_guard.set_native_tools( + set(_runtime._tool_gateway.registry.list_tools()) + ) + except Exception: + logger.warning( + "tool_guard: native-tool snapshot failed", exc_info=True + ) return _runtime diff --git a/server.py b/server.py index 6c1a714..31b8c50 100644 --- a/server.py +++ b/server.py @@ -4,6 +4,7 @@ import asyncio import json +import logging import os from pathlib import Path @@ -79,6 +80,8 @@ cancel_todo, ] +logger = logging.getLogger(__name__) + app = FastAPI(title="Roboot") STATIC_DIR = Path(__file__).parent / "static" @@ -130,6 +133,19 @@ def _get_runtime() -> arcana.Runtime: _runtime._tool_gateway.confirmation_callback = ( tool_guard.confirmation_callback ) + # Snapshot native tool names so tool_guard's side-effect-first + # gating treats only UNKNOWN/external (e.g. MCP) writes as + # gate-by-default — native low-risk writes (reminders/todos/notes/ + # voice) stay AUTO. Must run BEFORE any connect_mcp() so MCP tools + # are not captured as native. + try: + tool_guard.set_native_tools( + set(_runtime._tool_gateway.registry.list_tools()) + ) + except Exception: + logger.warning( + "tool_guard: native-tool snapshot failed", exc_info=True + ) return _runtime diff --git a/tests/test_tool_guard.py b/tests/test_tool_guard.py index 9e971e0..2148f31 100644 --- a/tests/test_tool_guard.py +++ b/tests/test_tool_guard.py @@ -46,12 +46,14 @@ def _isolate_paths(tmp_path, monkeypatch): def _reset_module_state(): tool_guard._broadcasters.clear() tool_guard._pending.clear() + tool_guard.set_native_tools(set()) # default: no natives → every write gates yield tool_guard._broadcasters.clear() for fut in list(tool_guard._pending.values()): if not fut.done(): fut.cancel() tool_guard._pending.clear() + tool_guard.set_native_tools(set()) # ----------------------------------------------------------------------------- @@ -744,3 +746,253 @@ async def bc(frame): assert "issued_at" in captured[0] assert isinstance(captured[0]["issued_at"], (int, float)) assert "timeout_s" in captured[0] + + +# ----------------------------------------------------------------------------- +# gate — side-effect-first gating (the MCP unknown-write path) +# +# Arcana only calls confirmation_callback for WRITE / requires_confirmation +# tools. An unknown WRITE tool (e.g. an MCP `gmail.send_email`) used to +# short-circuit to AUTO via the hardcoded name set — the gate-bypass hole +# (MCP tool-poisoning / unknown-WRITE-bypass, CVE-2025-54136 class). These +# tests pin the side-effect-first behavior that closes it. +# ----------------------------------------------------------------------------- + + +def test_coerce_side_effect_variants(): + coerce = tool_guard._coerce_side_effect + assert coerce(None) is None + assert coerce("write") == "write" + assert coerce("WRITE") == "write" + assert coerce(" Read ") == "read" + assert coerce("") is None + # Arcana's SideEffect is `class SideEffect(str, Enum)` — duck-typed via + # `.value` so tool_guard needs no arcana import. + assert coerce(SimpleNamespace(value="write")) == "write" + assert coerce(SimpleNamespace(value="NONE")) == "none" + + +async def test_gate_unknown_write_tool_is_gated_in_log(monkeypatch, _isolate_paths): + """An unknown tool flagged WRITE (e.g. MCP send_email) must be gated, not + auto-allowed. In LOG mode it lands in the audit with a synthesized reason + and the JSON-serialized args as the summary (no primary-text extractor).""" + monkeypatch.setenv("ROBOOT_TOOL_APPROVAL", "log") + decision = await tool_guard.gate( + "send_email", {"to": "x@y.z", "body": "hi"}, side_effect="write" + ) + assert decision == Decision.LOGGED + files = list(tool_guard.AUDIT_DIR.iterdir()) + assert len(files) == 1 + record = json.loads(files[0].read_text()) + assert record["tool"] == "send_email" + assert record["side_effect"] == "write" + assert "WRITE" in record["danger_reason"] + assert "x@y.z" in record["args_summary"] + + +async def test_gate_unknown_read_tool_is_not_gated(monkeypatch, _isolate_paths): + """An unknown READ tool stays AUTO — only writes/confirm are gated.""" + monkeypatch.setenv("ROBOOT_TOOL_APPROVAL", "confirm") + decision = await tool_guard.gate( + "search_inbox", {"q": "invoices"}, side_effect="read" + ) + assert decision == Decision.AUTO + # AUTO never writes audit, so the dir is never even created. + assert not (tool_guard.AUDIT_DIR.exists() and list(tool_guard.AUDIT_DIR.iterdir())) + + +async def test_gate_unknown_none_tool_is_not_gated(monkeypatch, _isolate_paths): + """side_effect='none' and unflagged → AUTO.""" + monkeypatch.setenv("ROBOOT_TOOL_APPROVAL", "confirm") + decision = await tool_guard.gate("ping", {}, side_effect="none") + assert decision == Decision.AUTO + + +async def test_gate_unknown_requires_confirmation_is_gated( + monkeypatch, _isolate_paths +): + """requires_confirmation forces gating even for a READ tool.""" + monkeypatch.setenv("ROBOOT_TOOL_APPROVAL", "log") + decision = await tool_guard.gate( + "read_clipboard", {}, side_effect="read", requires_confirmation=True + ) + assert decision == Decision.LOGGED + + +async def test_gate_unknown_write_tool_confirm_modal(monkeypatch, _isolate_paths): + """In CONFIRM the unknown write tool fires the modal and honors reject.""" + monkeypatch.setenv("ROBOOT_TOOL_APPROVAL", "confirm") + captured: list[dict] = [] + + async def bc(frame): + captured.append(frame) + tool_guard.resolve_decision(frame["req_id"], approved=False) + + tool_guard.register_broadcaster(bc) + decision = await tool_guard.gate( + "calendar_delete_event", {"id": "evt_1"}, side_effect="write" + ) + assert decision == Decision.REJECTED + assert captured and captured[0]["tool"] == "calendar_delete_event" + assert "WRITE" in captured[0]["danger_reason"] + + +async def test_gate_unknown_write_tool_cannot_be_allowlisted( + monkeypatch, _isolate_paths +): + """Unknown write tools have no primary-text extractor, so a prefix + allowlist entry can't waive their gate.""" + monkeypatch.setenv("ROBOOT_TOOL_APPROVAL", "log") + _write_allowlist( + tool_guard.ALLOWLIST_PATH, [{"tool": "send_email", "prefix": "x@y.z"}] + ) + decision = await tool_guard.gate( + "send_email", {"to": "x@y.z"}, side_effect="write" + ) + assert decision == Decision.LOGGED # still gated, allowlist didn't apply + + +async def test_gate_known_tool_precedence_over_side_effect( + monkeypatch, _isolate_paths +): + """shell is registered side_effect=write, but name-keyed danger-matching + wins: a safe shell command stays AUTO even though WRITE is passed (no + modal spam), while a dangerous one still gates.""" + monkeypatch.setenv("ROBOOT_TOOL_APPROVAL", "confirm") + safe = await tool_guard.gate( + "shell", {"command": "git status"}, side_effect="write" + ) + assert safe == Decision.AUTO + + captured: list[dict] = [] + + async def bc(frame): + captured.append(frame) + tool_guard.resolve_decision(frame["req_id"], approved=True) + + tool_guard.register_broadcaster(bc) + dangerous = await tool_guard.gate( + "shell", {"command": "rm -rf /tmp/x"}, side_effect="write" + ) + assert dangerous == Decision.APPROVED + assert "rm" in captured[0]["danger_reason"] + + +async def test_confirmation_callback_gates_mcp_write_tool( + monkeypatch, _isolate_paths +): + """End-to-end: a spec carrying SideEffect.WRITE drives the callback to gate + an unknown MCP tool and honor a reject.""" + monkeypatch.setenv("ROBOOT_TOOL_APPROVAL", "confirm") + + async def bc(frame): + tool_guard.resolve_decision(frame["req_id"], approved=False) + + tool_guard.register_broadcaster(bc) + tool_call = SimpleNamespace( + name="gmail.send_email", arguments={"to": "x@y.z", "subject": "hi"} + ) + spec = SimpleNamespace( + side_effect=SimpleNamespace(value="write"), requires_confirmation=False + ) + assert await tool_guard.confirmation_callback(tool_call, spec) is False + + +async def test_confirmation_callback_allows_mcp_read_tool(monkeypatch): + """A READ-classified MCP tool reaching the callback is not gated by us.""" + monkeypatch.setenv("ROBOOT_TOOL_APPROVAL", "confirm") + tool_call = SimpleNamespace(name="gmail.list_messages", arguments={}) + spec = SimpleNamespace( + side_effect=SimpleNamespace(value="read"), requires_confirmation=False + ) + assert await tool_guard.confirmation_callback(tool_call, spec) is True + + +async def test_confirmation_callback_failclosed_on_malformed_spec( + monkeypatch, _isolate_paths +): + """A spec missing BOTH side_effect and requires_confirmation must fail + CLOSED (treated as write) so an unknown tool gates, not slips by. (Not + reachable via Arcana's real contract, but defense-in-depth.)""" + monkeypatch.setenv("ROBOOT_TOOL_APPROVAL", "confirm") + + async def bc(frame): + tool_guard.resolve_decision(frame["req_id"], approved=False) + + tool_guard.register_broadcaster(bc) + tool_call = SimpleNamespace(name="mystery.write_thing", arguments={"x": 1}) + spec = SimpleNamespace() # no side_effect, no requires_confirmation + assert await tool_guard.confirmation_callback(tool_call, spec) is False + + +# ----------------------------------------------------------------------------- +# Native-vs-external trust boundary (set_native_tools) +# +# The side-effect-first path must gate UNKNOWN/external writes (MCP) WITHOUT +# re-gating Roboot's own low-risk native writes (reminders/todos/voice/notes), +# which were AUTO before. Native tools are exempt unless name-keyed or they +# explicitly declare requires_confirmation. +# ----------------------------------------------------------------------------- + + +async def test_native_write_exempt_external_write_gated(monkeypatch, _isolate_paths): + monkeypatch.setenv("ROBOOT_TOOL_APPROVAL", "confirm") + tool_guard.set_native_tools( + {"schedule_reminder", "add_todo", "switch_tts_voice", "update_self"} + ) + + captured: list[dict] = [] + + async def bc(frame): + captured.append(frame) + tool_guard.resolve_decision(frame["req_id"], approved=False) + + tool_guard.register_broadcaster(bc) + + # Native low-risk write → AUTO, no modal (no regression / no modal-spam). + native = await tool_guard.gate( + "schedule_reminder", + {"text": "买牛奶", "delay_seconds": 900}, + side_effect="write", + ) + assert native == Decision.AUTO + assert captured == [] + + # Unknown/external write (not native) → still gates. + external = await tool_guard.gate( + "gmail.send_email", {"to": "x@y.z"}, side_effect="write" + ) + assert external == Decision.REJECTED + assert captured and captured[0]["tool"] == "gmail.send_email" + + +async def test_native_tool_with_requires_confirmation_still_gates( + monkeypatch, _isolate_paths +): + """The native exemption must NOT swallow an explicit requires_confirmation + — an author who flags a native tool for confirmation still gets gated.""" + monkeypatch.setenv("ROBOOT_TOOL_APPROVAL", "log") + tool_guard.set_native_tools({"sensitive_native"}) + decision = await tool_guard.gate( + "sensitive_native", {}, side_effect="write", requires_confirmation=True + ) + assert decision == Decision.LOGGED + files = list(tool_guard.AUDIT_DIR.iterdir()) + assert len(files) == 1 + assert json.loads(files[0].read_text())["danger_reason"] == "requires_confirmation" + + +def test_create_claude_session_primary_text(): + """The _ALWAYS_CONFIRM entry is create_claude_session (not the dead + 'create_session'); its primary text is the prompt/dir for summary+allowlist.""" + assert ( + tool_guard._primary_text( + "create_claude_session", {"directory": "/p", "initial_prompt": "go"} + ) + == "go" + ) + assert ( + tool_guard._primary_text("create_claude_session", {"directory": "/p"}) == "/p" + ) + assert "create_claude_session" in tool_guard._ALWAYS_CONFIRM_TOOLS + assert "create_session" not in tool_guard._ALWAYS_CONFIRM_TOOLS diff --git a/tests/test_tool_guard_integration.py b/tests/test_tool_guard_integration.py index 5fd131c..162e54f 100644 --- a/tests/test_tool_guard_integration.py +++ b/tests/test_tool_guard_integration.py @@ -154,6 +154,56 @@ async def bc(frame): assert captured == [], "safe command should not have triggered the modal" +async def test_gateway_routes_external_write_tool_to_gate( + monkeypatch, _isolate_paths +): + """Pin Arcana's routing contract: the gateway invokes confirmation_callback + for a WRITE tool whose name is OUTSIDE Roboot's native/keyed set (the MCP + case), and our gate rejects it when the user says no. If a future Arcana + stops routing WRITE tools to the callback, this fails loudly — it's the + premise the unit-level MCP test takes on faith. + + Dispatches through the real `gateway.call()`, not the callback directly. + """ + monkeypatch.setenv("ROBOOT_TOOL_APPROVAL", "confirm") + + @arcana.tool( + when_to_use="stand-in for an external MCP write tool", + side_effect="write", + ) + async def fake_external_write(target: str) -> str: + return f"wrote {target}" + + rt = arcana.Runtime( + providers={"deepseek": "sk-fake"}, + tools=[fake_external_write], + budget=arcana.Budget(max_cost_usd=0.01), + config=arcana.RuntimeConfig(default_provider="deepseek"), + ) + rt._tool_gateway.confirmation_callback = tool_guard.confirmation_callback + # Deliberately do NOT call set_native_tools — an empty native set means an + # unrecognised write gates (fail-safe), which is exactly the MCP posture. + + denied: list[str] = [] + + async def deny(frame): + denied.append(frame["tool"]) + tool_guard.resolve_decision(frame["req_id"], approved=False) + + tool_guard.register_broadcaster(deny) + + call = ToolCall( + id="ext1", name="fake_external_write", arguments={"target": "prod"} + ) + result = await rt._tool_gateway.call(call) + + # The gateway routed the WRITE tool to our gate; the broadcaster rejected it, + # so the tool body never ran. + assert denied == ["fake_external_write"], "gateway did not route to the gate" + assert result.success is False + assert result.error is not None and result.error.code == "CONFIRMATION_REJECTED" + + async def test_off_mode_allows_dangerous(runtime, monkeypatch): """When ROBOOT_TOOL_APPROVAL is unset (default off), even dangerous commands pass — preserving back-compat for users who haven't opted in. diff --git a/tool_guard.py b/tool_guard.py index 87ed6a5..a57eb6a 100644 --- a/tool_guard.py +++ b/tool_guard.py @@ -2,11 +2,21 @@ Hooks into Arcana's `ToolGateway.confirmation_callback` so that any tool flagged `requires_confirmation=True` (or `side_effect=WRITE`) routes through -this module before execution. The gated set is `shell` (danger-pattern +this module before execution. The name-keyed set is `shell` (danger-pattern matched on the command), the iTerm write tools `send_to_session` / `create_session`, `enroll_face`, and the filesystem writers `write_file` / `edit_file` (always-confirm in CONFIRM mode, keyed on the target path, -allowlistable). Any other tool short-circuits to AUTO. +allowlistable). Roboot's OTHER native writes (reminders/todos/notes/voice) are +registered via `set_native_tools()` and stay AUTO — they were never in the +gated set and aren't now. But any UNKNOWN / external tool Arcana flags WRITE / +requires_confirmation — e.g. an MCP write tool such as `gmail.send_email`, +which is NOT a native tool — is gated by default (side-effect-first, not +name-allowlisted): without this an undeclared write tool reaching +`confirmation_callback` would short-circuit to AUTO and bypass approval +entirely (the tool-name-allowlist hole; the MCP tool-poisoning / +unknown-WRITE-bypass class, CVE-2025-54136). A tool that explicitly declares +`requires_confirmation=True` is always gated. Unknown READ/none tools that are +not flagged short-circuit to AUTO. Modes (env `ROBOOT_TOOL_APPROVAL`): off — bypass entirely; callback always allows (default). @@ -51,6 +61,30 @@ MAX_ARGS_BYTES = 2048 DEFAULT_TIMEOUT_S = 30.0 +# Tools gated by NAME. `shell` is DANGER-MATCHED — gated only when its command +# trips a danger pattern (safe commands fast-path to AUTO). The rest are +# ALWAYS-CONFIRM — every call is gated in CONFIRM mode (allowlistable by their +# primary text). These names are RESERVED for Roboot-native tools: the +# fast-path trusts `_primary_text` to know each tool's primary arg. An external +# tool can't collide (Arcana namespaces MCP tools as "server.tool"), but a +# future native tool must not reuse these names with a different arg shape. +_DANGER_MATCHED_TOOLS = {"shell"} +_ALWAYS_CONFIRM_TOOLS = { + "send_to_session", + "create_claude_session", + "enroll_face", + "write_file", + "edit_file", +} + +# Names of Roboot's OWN (native, vetted) tools — populated at startup via +# `set_native_tools()`. The side-effect-first path in `gate()` gates UNKNOWN / +# external WRITE tools (e.g. MCP) by default, but native low-risk writes +# (reminders/todos/notes/voice) must stay AUTO as they always were. Empty +# (never registered) fails safe: every write gates. Native tools still gate +# when name-keyed above, or when they explicitly set requires_confirmation. +_native_tools: set[str] = set() + class Mode(str, Enum): OFF = "off" @@ -271,8 +305,8 @@ def _primary_text(tool_name: str, args: dict) -> str: return str(args.get("command") or "") if tool_name == "send_to_session": return str(args.get("text") or "") - if tool_name == "create_session": - return str(args.get("command") or "") + if tool_name == "create_claude_session": + return str(args.get("initial_prompt") or args.get("directory") or "") if tool_name in ("write_file", "edit_file"): return str(args.get("path") or "") return "" @@ -343,6 +377,24 @@ def _log_audit(record: dict, suffix: str = "") -> Path | None: _pending: dict[str, asyncio.Future] = {} +def set_native_tools(names: set[str]) -> None: + """Register the names of Roboot's OWN (native, vetted) tools. + + The side-effect-first path in `gate()` gates UNKNOWN/external WRITE tools + (e.g. MCP) by default, but must NOT start gating Roboot's own low-risk + writes (reminders/todos/notes/voice) that were AUTO before. Native tools + listed here are exempt from the side-effect-first path; they still gate if + name-keyed (shell / iTerm / file writes) or if they explicitly declare + `requires_confirmation`. + + Call ONCE at startup, AFTER registering native tools but BEFORE any + `connect_mcp()`, so MCP tools are NOT captured as native. Empty set + (never registered) fails safe: every write gates. + """ + global _native_tools + _native_tools = set(names) + + def register_broadcaster(fn: Callable[[dict], Awaitable[None]]) -> None: if fn not in _broadcasters: _broadcasters.append(fn) @@ -386,40 +438,70 @@ async def gate( *, origin: str = "unknown", timeout: float = DEFAULT_TIMEOUT_S, + side_effect: str | None = None, + requires_confirmation: bool = False, ) -> Decision: """The single approval entry point. + `side_effect` ("write"/"read"/"none") and `requires_confirmation` come from + the tool's Arcana spec via `confirmation_callback`. Direct callers (tests, + in-process) may omit them — name-keyed tools gate the same way they always + have; only UNKNOWN tools change behaviour, and only when flagged WRITE. + Logic (precedence matters — danger always wins over allowlist): 1. mode=OFF → AUTO. - 2. unknown tool → AUTO (only the v1 gated set is policed). - 3. tool=shell + no danger pattern → AUTO (the common case, fast path). - 4. always-confirm tool (non-shell) + allowlist hit → AUTO. - shell + danger → allowlist CANNOT override; falls through to modal. - 5. summary > MAX_ARGS_BYTES → REJECTED. - 6. mode=LOG (or no broadcasters in CONFIRM) → LOGGED + audit + allow. - 7. mode=CONFIRM → broadcast frame, await reply or timeout. + 2. Gating eligibility (precedence: native name-keyed policy → explicit + confirmation → side-effect-first for unknown/external writes → AUTO): + - name in _DANGER_MATCHED_TOOLS (shell): gate ONLY if the command + trips a danger pattern; safe commands fast-path to AUTO. + - name in _ALWAYS_CONFIRM_TOOLS: every call gated. + - requires_confirmation set: gated (explicit author request). + - side_effect=write AND not a native tool (e.g. an MCP write): + gated by default — closes the unknown/external-write bypass. + - else (native low-risk write, unknown READ/none, unflagged): AUTO. + 3. no danger + allowlist hit → AUTO. + A dangerous shell command CANNOT be allowlisted; falls through to modal. + 4. summary > MAX_ARGS_BYTES → REJECTED. + 5. mode=LOG (or no broadcasters in CONFIRM) → LOGGED + audit + allow. + 6. mode=CONFIRM → broadcast frame, await reply or timeout. """ primary = _primary_text(tool_name, args) - danger = detect_dangerous(primary) if tool_name == "shell" else None + danger = detect_dangerous(primary) if tool_name in _DANGER_MATCHED_TOOLS else None mode = get_mode() if mode == Mode.OFF: return Decision.AUTO - if tool_name not in { - "shell", - "send_to_session", - "create_session", - "enroll_face", - "write_file", - "edit_file", - }: - # Unknown tool — don't gate. Future tools opt in. - return Decision.AUTO - if tool_name == "shell" and danger is None: + + se = (side_effect or "").strip().lower() + + # Gating eligibility — native name-keyed policy first, then side-effect-first. + reason = danger + if tool_name in _DANGER_MATCHED_TOOLS: + # shell: gate ONLY dangerous commands. Safe ones fast-path to AUTO even + # though shell is side_effect=write+requires_confirmation — we don't + # modal-spam every `ls`. + if danger is None: + return Decision.AUTO + elif tool_name in _ALWAYS_CONFIRM_TOOLS: + pass # native high-risk write — every call gated (allowlistable below) + elif requires_confirmation: + # Explicitly flagged by the tool author — gate whether native or not. + reason = reason or "requires_confirmation" + elif se == "write" and tool_name not in _native_tools: + # UNKNOWN / external WRITE — e.g. an MCP write tool (gmail.send_email). + # Arcana only reaches this callback for WRITE/requires_confirmation + # tools, so an unrecognised one is an external action worth approving. + # Native low-risk writes are in _native_tools and fall through to AUTO. + reason = f"unknown WRITE tool (side_effect={se or 'inferred'})" + else: + # Native low-risk write, or unknown READ/none, or unflagged — don't gate. return Decision.AUTO + # Allowlist applies ONLY when there is no danger. A dangerous shell # command must always go to modal — the allowlist is for suppressing # routine confirmations on always-confirm tools, not for waiving danger. + # Unknown WRITE tools have no `_primary_text` extractor, so `is_allowlisted` + # returns False for them — they cannot be prefix-allowlisted today. if danger is None and is_allowlisted(tool_name, args): return Decision.AUTO @@ -427,7 +509,8 @@ async def gate( record_base = { "tool": tool_name, "args_summary": summary, - "danger_reason": danger, + "danger_reason": reason, + "side_effect": se or None, "origin": origin, "ts": time.time(), } @@ -461,7 +544,7 @@ async def gate( "req_id": req_id, "tool": tool_name, "args_summary": summary, - "danger_reason": danger, + "danger_reason": reason, "origin": origin, "issued_at": time.time(), "timeout_s": timeout, @@ -507,6 +590,19 @@ async def gate( ) +def _coerce_side_effect(side_effect: Any) -> str | None: + """Normalize Arcana's `SideEffect` enum (or a raw string) to its lowercase + value ('write'/'read'/'none'), or None if absent. + + tool_guard deliberately carries no `arcana` import, so we duck-type the + enum's `.value` rather than compare against `SideEffect.WRITE`. + """ + if side_effect is None: + return None + val = getattr(side_effect, "value", side_effect) + return str(val).strip().lower() or None + + async def confirmation_callback(tool_call: Any, spec: Any) -> bool: """Adapter for `arcana.tool_gateway.gateway.ToolGateway.confirmation_callback`. @@ -532,7 +628,20 @@ async def confirmation_callback(tool_call: Any, spec: Any) -> bool: if not isinstance(raw_args, dict): raw_args = {} origin = current_origin.get() - decision = await gate(name, raw_args, origin=origin) + side_effect = _coerce_side_effect(getattr(spec, "side_effect", None)) + requires_confirmation = bool(getattr(spec, "requires_confirmation", False)) + # Arcana only invokes this callback for WRITE / requires_confirmation + # tools. If the spec is malformed and we can read NEITHER signal, fail + # closed — treat it as a write so an unknown tool gates, not slips by. + if side_effect is None and not requires_confirmation: + side_effect = "write" + decision = await gate( + name, + raw_args, + origin=origin, + side_effect=side_effect, + requires_confirmation=requires_confirmation, + ) return decision in {Decision.AUTO, Decision.LOGGED, Decision.APPROVED} except Exception: logger.exception("tool_guard: gate crashed; failing closed") diff --git a/tools/vision.py b/tools/vision.py index 069c07f..44c6263 100644 --- a/tools/vision.py +++ b/tools/vision.py @@ -173,6 +173,10 @@ async def look() -> str: when_to_use="当用户说'记住我'、'记住我的脸'、'注册人脸'、'我是xxx'，或让你认识一个新面孔时", what_to_expect="拍照并把用户的脸存入数据库，下次就能认出来", failure_meaning="拍照失败或没检测到人脸", + # Writes a face encoding to the .faces DB. side_effect="write" makes Arcana + # invoke tool_guard.confirmation_callback so this is actually gated in + # CONFIRM mode (it's listed in tool_guard._ALWAYS_CONFIRM_TOOLS). + side_effect="write", ) async def enroll_face(name: str) -> str: """拍照并注册人脸。name 是这个人的名字。会拍多张提高准确度。"""