From 3f2dcc6c11106cf949c94806a3ba3e65223b0feb Mon Sep 17 00:00:00 2001 From: William Chastain Date: Sat, 1 Aug 2026 20:31:51 -0700 Subject: [PATCH 1/2] feat(gate): argument-conditional approval cards (gate.ask_when) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit One tool can be two actions: hound's smart_fetch reads a page, but the same call carrying `actions` clicks and submits on it. The gate could only decide on a tool name, so the owner had to either approve every page read or accept that approving fetch also approved clicking. `gate.ask_when` maps a tool to argument names that pull it back into the card path however it was approved — it outranks `"*"`, an explicit approve, and the read-only fast path; `gate.never` still wins. Matching is on presence, not value. "always" stays available on these cards like any other, but persists `tool:argument` rather than the bare name, so one tap never approves the tool's other watched arguments. Splits GatePolicy and the approved-set persistence into gate_policy.py — gate.py was at the 200-line cap, and the decision rules and the enforcement path are separate concerns. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01EP49QtT1zxHvm1PPCmCpUz --- docs/CONFIG.md | 19 +++++- src/chief/config/coerce.py | 17 ++++++ src/chief/config/load.py | 1 + src/chief/config/schema.py | 4 ++ src/chief/gate.py | 82 ++++++++----------------- src/chief/gate_policy.py | 97 ++++++++++++++++++++++++++++++ src/chief/wiring.py | 9 +-- tests/test_config.py | 23 +++++++ tests/test_gate.py | 120 +++++++++++++++++++++++++++++++++++-- 9 files changed, 302 insertions(+), 70 deletions(-) create mode 100644 src/chief/gate_policy.py diff --git a/docs/CONFIG.md b/docs/CONFIG.md index 84490af3..97f5f8cc 100644 --- a/docs/CONFIG.md +++ b/docs/CONFIG.md @@ -55,6 +55,7 @@ template line. | *(secret file)* | `openrouter_api_key` | `str` | `""` | `OPENROUTER_API_KEY` | | `gate.never` | `gate_never` | `tuple` | `()` | — | | `gate.approved` | `gate_approved` | `tuple` | `()` | — | +| `gate.ask_when` | `gate_ask_when` | `dict` | `{}` | — | | `gate.announce` | `gate_announce` | `bool` | `True` | — | | `budget.cap_usd` | `budget_cap_usd` | `float` | `0.0` (unlimited) | — | | `budget.warn_ratio` | `budget_warn_ratio` | `float` | `0.8` | — | @@ -95,6 +96,22 @@ Notes on specific keys: env first, then `secrets/`. Empty `web_password` means the web UI fails closed and no listener is built. - **`gate.approved` accepts `"*"`** to approve every tool; `gate.never` still wins. +- **`gate.ask_when` cards an approved tool when a named argument is present** — + for tools that are two actions in one. hound's `smart_fetch` reads a page, but + the same call carrying `actions` clicks and submits on it: + + ```yaml + gate: + approved: [mcp_hound_smart_fetch] + ask_when: + mcp_hound_smart_fetch: [actions] + ``` + + Matching is on presence, not value, and it outranks every auto-approve — + `"*"`, an explicit approve, and the read-only fast path. `gate.never` still + wins. Answering **always** to such a card persists `mcp_hound_smart_fetch:actions` + (tool + the argument that raised it) rather than the bare tool name, so one + tap never approves the tool's other watched arguments. - **`imessage.enabled` also requires `sys.platform == "darwin"`.** - **`imessage.mode` picks which Apple ID chief speaks as** — `self` (default, today's install: the owner's own, chief texted through the self-chat) or @@ -164,7 +181,7 @@ hand-edit config from a package install; use `config_apply`. | `chief.db` | sqlite sessions + transcripts; schema created at boot, **no migrations** | | `chief.sock` | unix socket for `chief-cli` | | `audit.jsonl` | every gated tool call | -| `gate_approved.json` | persisted "always allow" set, minus config-approved | +| `gate_approved.json` | persisted "always allow" set, minus config-approved; a `tool:argument` entry is an `ask_when` grant | | `installed.yaml` | package install registry | | `packages/` | clone of `packages_repo`, pulled by `chief-pkg update` | | `hooks//` | per-package hook scratch state | diff --git a/src/chief/config/coerce.py b/src/chief/config/coerce.py index 6ebbfa41..37cfe489 100644 --- a/src/chief/config/coerce.py +++ b/src/chief/config/coerce.py @@ -45,6 +45,23 @@ def as_handles(value: Any) -> tuple[str, ...]: ) +def ask_when(raw: Any) -> dict[str, tuple[str, ...]]: + """Coerce ``gate.ask_when`` to ``tool name -> watched argument names``. + + Values take the same shapes :func:`as_handles` accepts, so a single + argument needs no list (``fetch: actions``). + """ + if not raw: + return {} + if not isinstance(raw, dict): + raise ConfigError( + "gate.ask_when must be a mapping of tool name -> argument names " + f"(e.g. {{mcp_hound_smart_fetch: [actions]}}), got " + f"{type(raw).__name__} {raw!r}" + ) + return {str(tool): as_handles(args) for tool, args in raw.items()} + + def autonomy(value: Any) -> str: """Coerce ``update.autonomy``; reject anything outside the three values. diff --git a/src/chief/config/load.py b/src/chief/config/load.py index 40312bc4..29a05df7 100644 --- a/src/chief/config/load.py +++ b/src/chief/config/load.py @@ -103,6 +103,7 @@ def load_config(path: Path = Path("config.yaml")) -> Config: provider_aliases=coerce.aliases(raw.get("provider_aliases") or {}), gate_never=tuple(gate.get("never") or ()), gate_approved=tuple(gate.get("approved") or ()), + gate_ask_when=coerce.ask_when(gate.get("ask_when")), gate_announce=bool(gate.get("announce", True)), budget_cap_usd=float(budget.get("cap_usd", 0.0)), budget_warn_ratio=float(budget.get("warn_ratio", 0.8)), diff --git a/src/chief/config/schema.py b/src/chief/config/schema.py index 01f57e62..a082c831 100644 --- a/src/chief/config/schema.py +++ b/src/chief/config/schema.py @@ -72,6 +72,10 @@ class Config: provider_aliases: dict[str, "AliasSpec"] = field(default_factory=dict) gate_never: tuple[str, ...] = () gate_approved: tuple[str, ...] = () + # Tool name -> argument names that force an approval card even when the + # tool is approved, for tools that are two actions in one (hound's + # smart_fetch reads a page; the same call carrying `actions` clicks on it). + gate_ask_when: dict[str, tuple[str, ...]] = field(default_factory=dict) # Announce every non-card tool call on the session's own surface, so an # approved / "always allow"ed tool stays visible instead of silent. gate_announce: bool = True diff --git a/src/chief/gate.py b/src/chief/gate.py index 1093c7a5..bd6a3c91 100644 --- a/src/chief/gate.py +++ b/src/chief/gate.py @@ -3,22 +3,22 @@ NEVER and APPROVED lists decide most calls; read-only tools auto-approve; the remaining gray zone raises an approval card on the session's own surface. From a card the owner can approve once or "always allow" a tool, which persists it -to the approved set (#187). Every call that does *not* raise a card is instead -announced on the session's surface as it starts, so an approved or "always" -tool stays visible to the owner instead of running silently. Everything -behavioral lives in prompts — this file only enforces. +to the approved set (#187). A ``gate.ask_when`` argument pulls an otherwise +approved tool back into the card path — see :mod:`chief.gate_policy` for the +rules and for what "always" persists there. Every call that does *not* raise a +card is instead announced on the session's surface as it starts, so an approved +or "always" tool stays visible to the owner instead of running silently. +Everything behavioral lives in prompts — this file only enforces. """ import json import logging from collections.abc import Awaitable, Callable -from dataclasses import dataclass, field -from enum import Enum -from pathlib import Path from chief.approvals import Approval, ApprovalBroker from chief.audit import AuditLog from chief.dispatch import WEB_CHANNEL, Dispatcher +from chief.gate_policy import Decision, GatePolicy, grant_key from chief.provider.base import ToolCall, ToolSpec from chief.tools import ToolContext, ToolDispatcher @@ -29,37 +29,6 @@ ANNOUNCE_ARG_LIMIT = 160 -class Decision(Enum): - NEVER = "never" - APPROVED = "approved" - ASK = "ask" - - -@dataclass -class GatePolicy: - """The code-enforced lists; anything on neither list asks. - - ``approved`` is a live set — an "always allow" answer adds to it so the - tool stops asking for the rest of the process (and is persisted so it - survives a restart). A ``"*"`` entry in ``approved`` matches every tool - name (the config-driven "all tools" switch); ``never`` still takes - precedence over it. - """ - - never: frozenset[str] = frozenset() - approved: set[str] = field(default_factory=set) - - def decide(self, tool_name: str) -> Decision: - if tool_name in self.never: - return Decision.NEVER - if "*" in self.approved or tool_name in self.approved: - return Decision.APPROVED - return Decision.ASK - - def allow_always(self, tool_name: str) -> None: - self.approved.add(tool_name) - - AskApproval = Callable[[ToolContext, str], Awaitable[Approval]] AllowAlways = Callable[[str], None] Announce = Callable[[ToolContext, str], Awaitable[None]] @@ -100,19 +69,6 @@ def announce_text(call: ToolCall, *, denied: bool = False) -> str: return f"⚙ {call.name} {arguments}{suffix}" -def load_approved(path: Path) -> set[str]: - """Read the persisted "always allow" tool names (empty if absent).""" - if not path.exists(): - return set() - return set(json.loads(path.read_text())) - - -def save_approved(names: set[str], path: Path) -> None: - """Persist the "always allow" tool names, sorted for a stable file.""" - path.parent.mkdir(parents=True, exist_ok=True) - path.write_text(json.dumps(sorted(names))) - - class GatedTools: """Per-session tool dispatcher: gate + audit around the shared registry.""" @@ -148,14 +104,17 @@ async def dispatch( # self-correct. Nothing runs, so nothing is announced either. self._record(call, "unknown_tool") return await self._registry.dispatch(call, self._context) - decision = self._policy.decide(call.name) - if decision is Decision.ASK and call.name in self._read_only(): + decision = self._policy.decide(call.name, call.arguments) + # A watched argument outranks every auto-approve, read_only included: + # the tool may only read, but the argument is what makes it act. + pending = self._policy.pending_arguments(call.name, call.arguments) + if decision is Decision.ASK and not pending and call.name in self._read_only(): decision = Decision.APPROVED self._record(call, "read_only") await self._announce_call(call, denied=False) elif decision is Decision.ASK: # The card already shows the call; a second line would double it. - decision = await self._ask_card(call) + decision = await self._ask_card(call, pending) else: self._record(call, f"list:{decision.value}") await self._announce_call(call, denied=decision is Decision.NEVER) @@ -177,11 +136,20 @@ async def _announce_call(self, call: ToolCall, *, denied: bool) -> None: except Exception: logger.exception("failed to announce tool call %s", call.name) - async def _ask_card(self, call: ToolCall) -> Decision: - question = f"approve tool call {call.name}({call.arguments})? yes / always / no" + async def _ask_card( + self, call: ToolCall, pending: tuple[str, ...] = () + ) -> Decision: + carries = f" — carries {', '.join(pending)}" if pending else "" + question = ( + f"approve tool call {call.name}({call.arguments})" + f"{carries}? yes / always / no" + ) answer = await self._ask(self._context, question) if answer is Approval.ALWAYS: - self._on_always(call.name) + # An ask_when card grants the argument that raised it, not the + # whole tool — otherwise one tap would approve every other use. + for grant in [grant_key(call.name, arg) for arg in pending] or [call.name]: + self._on_always(grant) self._record(call, f"card:{answer.value}") return Decision.NEVER if answer is Approval.DENY else Decision.APPROVED diff --git a/src/chief/gate_policy.py b/src/chief/gate_policy.py new file mode 100644 index 00000000..d201e25a --- /dev/null +++ b/src/chief/gate_policy.py @@ -0,0 +1,97 @@ +"""What the gate's lists decide, and how "always allow" is persisted. + +Split from :mod:`chief.gate` so that file stays the enforcement path (cards, +announcements, audit) and this one holds the decision rules and their storage. + +Two stores feed one effective approved set: ``gate.approved`` in config.yaml is +the owner's declared intent, and ``gate_approved.json`` accretes "always allow" +answers. They are unioned at boot (:func:`chief.wiring.build_gate`), which is +what lets a tap take effect without ever rewriting the owner's config. +""" + +import json +from collections.abc import Mapping, Sequence +from dataclasses import dataclass, field +from enum import Enum +from pathlib import Path +from typing import Any + +# An argument-scoped grant is one approved-set entry, so "always allow" on an +# ask_when card persists through the same file and the same code path as any +# other. Tool names are identifiers or "mcp__", so ":" cannot +# collide with a bare name. +GRANT_SEPARATOR = ":" + + +def grant_key(tool_name: str, argument: str) -> str: + """The approved-set entry standing for "this tool, carrying this argument".""" + return f"{tool_name}{GRANT_SEPARATOR}{argument}" + + +class Decision(Enum): + NEVER = "never" + APPROVED = "approved" + ASK = "ask" + + +@dataclass +class GatePolicy: + """The code-enforced lists; anything on neither list asks. + + ``approved`` is a live set — an "always allow" answer adds to it so the + tool stops asking for the rest of the process (and is persisted so it + survives a restart). A ``"*"`` entry in ``approved`` matches every tool + name (the config-driven "all tools" switch); ``never`` still takes + precedence over it. + + ``ask_when`` maps a tool name to argument names that pull it back into the + card path *however* it was approved. It exists because one tool can be two + actions: hound's ``smart_fetch`` reads a page, but the same call carrying + ``actions`` clicks and submits on it. Answering "always" to such a card + grants ``tool:argument`` (see :func:`grant_key`) rather than the bare name, + so the owner never has to approve the whole tool to allow one argument. + """ + + never: frozenset[str] = frozenset() + approved: set[str] = field(default_factory=set) + ask_when: Mapping[str, Sequence[str]] = field(default_factory=dict) + + def decide( + self, tool_name: str, arguments: Mapping[str, Any] | None = None + ) -> Decision: + if tool_name in self.never: + return Decision.NEVER + if self.pending_arguments(tool_name, arguments): + return Decision.ASK + if "*" in self.approved or tool_name in self.approved: + return Decision.APPROVED + return Decision.ASK + + def pending_arguments( + self, tool_name: str, arguments: Mapping[str, Any] | None = None + ) -> tuple[str, ...]: + """Watched arguments this call carries that have no standing grant. + + Non-empty means the call must be carded no matter what the lists say. + """ + watched = self.ask_when.get(tool_name) or () + present = (name for name in watched if name in (arguments or {})) + return tuple( + name for name in present if grant_key(tool_name, name) not in self.approved + ) + + def allow_always(self, tool_name: str) -> None: + self.approved.add(tool_name) + + +def load_approved(path: Path) -> set[str]: + """Read the persisted "always allow" entries (empty if absent).""" + if not path.exists(): + return set() + return set(json.loads(path.read_text())) + + +def save_approved(names: set[str], path: Path) -> None: + """Persist the "always allow" entries, sorted for a stable file.""" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(sorted(names))) diff --git a/src/chief/wiring.py b/src/chief/wiring.py index 36a4f9ec..a61f0b14 100644 --- a/src/chief/wiring.py +++ b/src/chief/wiring.py @@ -23,13 +23,9 @@ from chief.config import Config from chief.cron.service import CronService from chief.dispatch import Dispatcher -from chief.gate import GatePolicy, load_approved, save_approved +from chief.gate_policy import GatePolicy, load_approved, save_approved from chief.hub import ObserverHub -from chief.mcpclient.manager import ( - McpManager, - ServerConfig, - server_config_from_entry, -) +from chief.mcpclient.manager import McpManager, ServerConfig, server_config_from_entry from chief.monitors.service import MonitorService from chief.persistence.db import ( SessionFactory, @@ -134,6 +130,7 @@ def build_gate(config: Config) -> Gate: policy = GatePolicy( never=frozenset(config.gate_never), approved=config_approved | load_approved(approved_path), + ask_when=config.gate_ask_when, ) def allow_always(tool_name: str) -> None: diff --git a/tests/test_config.py b/tests/test_config.py index 38a71f73..faec2a6a 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -35,6 +35,29 @@ def test_shell_timeout_yaml_override(tmp_path: Path) -> None: assert load_config(path).shell_timeout_seconds == 45.0 +def test_gate_ask_when_loads_per_tool_argument_names(tmp_path: Path) -> None: + path = tmp_path / "config.yaml" + path.write_text( + "gate:\n" + " approved: [mcp_hound_smart_fetch]\n" + " ask_when:\n" + " mcp_hound_smart_fetch: [actions]\n" + ) + config = load_config(path) + assert config.gate_approved == ("mcp_hound_smart_fetch",) + assert config.gate_ask_when == {"mcp_hound_smart_fetch": ("actions",)} + + +def test_gate_ask_when_accepts_a_bare_string_argument(tmp_path: Path) -> None: + path = tmp_path / "config.yaml" + path.write_text("gate:\n ask_when:\n fetch: actions\n") + assert load_config(path).gate_ask_when == {"fetch": ("actions",)} + + +def test_gate_ask_when_defaults_empty(tmp_path: Path) -> None: + assert load_config(tmp_path / "missing.yaml").gate_ask_when == {} + + def test_yaml_values_override_defaults(tmp_path: Path) -> None: path = tmp_path / "config.yaml" path.write_text( diff --git a/tests/test_gate.py b/tests/test_gate.py index a04209f7..5f225d09 100644 --- a/tests/test_gate.py +++ b/tests/test_gate.py @@ -9,11 +9,11 @@ from chief.approvals import Approval, ApprovalBroker from chief.audit import AuditLog from chief.dispatch import Dispatcher -from chief.gate import ( +from chief.gate import GatedTools, approval_asker +from chief.gate_policy import ( Decision, - GatedTools, GatePolicy, - approval_asker, + grant_key, load_approved, save_approved, ) @@ -53,16 +53,58 @@ def test_star_wildcard_approves_every_tool() -> None: assert policy.decide("rm_rf") is Decision.NEVER +def ask_when_policy(*grants: str) -> GatePolicy: + """``peek`` is read-only and ``echo`` approved — both watch ``actions``.""" + return GatePolicy( + never=frozenset({"rm_rf"}), + approved={"echo", *grants}, + ask_when={"echo": ("actions",), "peek": ("actions", "fill")}, + ) + + +def test_ask_when_argument_cards_an_otherwise_approved_tool() -> None: + policy = ask_when_policy() + assert policy.decide("echo", {"text": "x"}) is Decision.APPROVED + assert policy.decide("echo", {"actions": [{"click": "a"}]}) is Decision.ASK + + +def test_ask_when_beats_the_star_wildcard() -> None: + policy = GatePolicy( + never=frozenset({"rm_rf"}), + approved={"*"}, + ask_when={"fetch": ("actions",)}, + ) + assert policy.decide("fetch", {"url": "u"}) is Decision.APPROVED + assert policy.decide("fetch", {"actions": []}) is Decision.ASK + assert policy.decide("rm_rf", {"actions": []}) is Decision.NEVER + + +def test_argument_scoped_grant_stops_the_carding() -> None: + policy = ask_when_policy(grant_key("echo", "actions")) + assert policy.decide("echo", {"actions": []}) is Decision.APPROVED + + +def test_grant_covers_only_the_argument_it_names() -> None: + """Granting ``actions`` must not silently approve a sibling argument.""" + policy = GatePolicy( + approved={"peek", grant_key("peek", "actions")}, + ask_when={"peek": ("actions", "fill")}, + ) + assert policy.decide("peek", {"actions": []}) is Decision.APPROVED + assert policy.decide("peek", {"actions": [], "fill": "x"}) is Decision.ASK + + def make_gated( tmp_path: Path, answer: Approval, *, on_always: object = None, announced: list[str] | None = None, + policy: GatePolicy | None = None, ) -> tuple[GatedTools, list[str]]: registry = ToolRegistry() - async def echo(text: str = "") -> str: + async def echo(text: str = "", **_ignored: object) -> str: return f"ran:{text}" specs = [ @@ -85,7 +127,7 @@ async def announce(context: ToolContext, text: str) -> None: gated = GatedTools( registry=registry, - policy=make_policy(), + policy=policy or make_policy(), audit=AuditLog(tmp_path / "audit.jsonl"), context=CONTEXT, ask=ask, @@ -141,6 +183,72 @@ async def test_always_answer_runs_and_persists_tool(tmp_path: Path) -> None: assert promoted == ["gray"] +async def test_ask_when_beats_the_read_only_auto_approve(tmp_path: Path) -> None: + """``peek`` is read-only, but a watched argument still has to be carded.""" + gated, questions = make_gated( + tmp_path, Approval.ONCE, policy=ask_when_policy() + ) + result = await gated.dispatch( + ToolCall(id="1", name="peek", arguments={"actions": [{"click": "a"}]}) + ) + assert result == "ran:" + assert len(questions) == 1 + assert "actions" in questions[0] + + +async def test_read_only_tool_still_auto_approves_without_the_argument( + tmp_path: Path, +) -> None: + gated, questions = make_gated( + tmp_path, Approval.DENY, policy=ask_when_policy() + ) + assert await gated.dispatch(ToolCall(id="1", name="peek", arguments={})) == "ran:" + assert questions == [] + + +async def test_always_on_an_ask_when_card_grants_only_that_argument( + tmp_path: Path, +) -> None: + """The switch stays consistent with every other card — it just persists + the tool+argument the rule named, not the tool wholesale.""" + promoted: list[str] = [] + gated, _ = make_gated( + tmp_path, + Approval.ALWAYS, + on_always=promoted.append, + policy=ask_when_policy(), + ) + await gated.dispatch( + ToolCall(id="1", name="echo", arguments={"actions": [], "text": "x"}) + ) + assert promoted == ["echo:actions"] + + +async def test_always_on_an_ordinary_card_still_grants_the_bare_name( + tmp_path: Path, +) -> None: + promoted: list[str] = [] + gated, _ = make_gated( + tmp_path, + Approval.ALWAYS, + on_always=promoted.append, + policy=ask_when_policy(), + ) + await gated.dispatch(ToolCall(id="1", name="gray", arguments={})) + assert promoted == ["gray"] + + +async def test_never_still_wins_over_a_watched_argument(tmp_path: Path) -> None: + gated, questions = make_gated( + tmp_path, Approval.ONCE, policy=ask_when_policy() + ) + result = await gated.dispatch( + ToolCall(id="1", name="rm_rf", arguments={"actions": []}) + ) + assert result == "error: tool 'rm_rf' denied by the gate" + assert questions == [] + + async def test_unknown_tool_errors_without_card(tmp_path: Path) -> None: """A phantom tool name never raises a card and never persists (audit C2).""" promoted: list[str] = [] @@ -230,7 +338,7 @@ async def test_announce_failure_does_not_break_the_call(tmp_path: Path) -> None: """A dead channel must not turn a working tool call into an error.""" registry = ToolRegistry() - async def echo(text: str = "") -> str: + async def echo(text: str = "", **_ignored: object) -> str: return f"ran:{text}" registry.register( From 5b4b3f733451e6a32081396cac83b8330a42990b Mon Sep 17 00:00:00 2001 From: William Chastain Date: Sun, 2 Aug 2026 11:42:05 -0700 Subject: [PATCH 2/2] fix(gate): a tool:argument grant approves on its own MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review high-1 on #296, owner-approved. `decide` checked `pending_arguments` (which returns () once every carried watched argument holds a grant), then fell through to the bare-name check. So a composite grant only ever lifted the ask_when veto — it never approved. For a tool not independently in `approved` / `"*"` / `read_only`, answering "always" to an ask_when card was a no-op that re-carded forever: p = GatePolicy(approved={'peek:actions'}, ask_when={'peek': ('actions',)}) p.pending_arguments('peek', {'actions': []}) # () p.decide('peek', {'actions': []}) # was ASK, now APPROVED A call carrying a watched argument is now decided by its grants alone, before the bare-name branch: all granted → APPROVED, any pending → ASK. The bare name is neither required nor sufficient there, so the grant stays narrow — it never approves a call without that argument, and never covers a sibling argument. Both existing policy tests seeded the bare name alongside the grant, which is why this slipped through; the two new tests do not. Docs the review also flagged: - SECURITY.md documented `decide(name)` with no ask_when step at all. Rewritten with the real order, the read_only override, and why grants are checked first. - CONFIG.md never said that presence-only matching makes one "always" cover every future value of the argument, nor that the argument name is unvalidated against the tool's schema (a typo silently disables the rule). Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01EP49QtT1zxHvm1PPCmCpUz --- docs/CONFIG.md | 12 ++++++++++++ docs/SECURITY.md | 31 +++++++++++++++++++++++-------- src/chief/gate_policy.py | 24 +++++++++++++++++++----- tests/test_gate.py | 33 +++++++++++++++++++++++++++++++++ 4 files changed, 87 insertions(+), 13 deletions(-) diff --git a/docs/CONFIG.md b/docs/CONFIG.md index 97f5f8cc..5c60597b 100644 --- a/docs/CONFIG.md +++ b/docs/CONFIG.md @@ -112,6 +112,18 @@ Notes on specific keys: wins. Answering **always** to such a card persists `mcp_hound_smart_fetch:actions` (tool + the argument that raised it) rather than the bare tool name, so one tap never approves the tool's other watched arguments. + + Two consequences of matching on presence alone. **One "always" covers every + future value of that argument** — granting `actions` once approves every later + call carrying `actions`, whatever it clicks; the gate does not read what is + inside. And the grant *is* the approval: a tool that is otherwise unapproved + becomes callable **only** for calls carrying a granted argument, and still + cards for anything else. + + The argument name is matched verbatim against the call's arguments and is + **not validated against the tool's schema** — a typo names an argument that + never arrives, silently leaving the tool on whatever its ordinary listing + says. Check the spelling against the tool's parameters. - **`imessage.enabled` also requires `sys.platform == "darwin"`.** - **`imessage.mode` picks which Apple ID chief speaks as** — `self` (default, today's install: the owner's own, chief texted through the self-chat) or diff --git a/docs/SECURITY.md b/docs/SECURITY.md index d9abc3d4..217f73c1 100644 --- a/docs/SECURITY.md +++ b/docs/SECURITY.md @@ -24,25 +24,40 @@ Three sender values matter (`dispatch.py`): `owner`, `system`, and anything else ## The gate — `gate.py` -**Gating is by tool name only.** Arguments are never inspected for the decision; -they are only rendered into announcement and card text. A tool approved once with -"always" is approved for every argument it will ever receive — an approved `shell` -is an approved *anything*. +**Gating is by tool name, plus `ask_when`'s argument *presence*.** Argument +*values* are never inspected for the decision; they are only rendered into +announcement and card text. Outside an `ask_when` rule, a tool approved once +with "always" is approved for every argument it will ever receive — an approved +`shell` is an approved *anything*. `GatedTools.dispatch` decides in strict order: 1. **Unknown tool name** → record `unknown_tool`, pass through to the registry for a helpful error. **A hallucinated name must never raise a card**, and must never be able to persist an "always" for a tool that doesn't exist. -2. **`policy.decide(name)`** — `never` → NEVER; `"*"` or an exact name in - `approved` → APPROVED; else ASK. **`never` is checked first and beats the `"*"` - wildcard.** -3. **ASK + `read_only`** → auto-approved and announced. +2. **`policy.decide(name, arguments)`** — `never` → NEVER. Then, if the call + carries any argument named by that tool's `ask_when` rule, **the call is + decided by its grants alone**: every carried watched argument holding a + `tool:argument` grant → APPROVED, otherwise ASK. The bare name is neither + required nor sufficient there. Only when no watched argument is present do + `"*"` or an exact name in `approved` → APPROVED; else ASK. **`never` is + checked first and beats everything; `ask_when` beats `"*"` and an explicit + approve.** +3. **ASK + `read_only`** → auto-approved and announced — *unless* the call has + pending watched arguments. **`ask_when` outranks `read_only`**: the tool may + only read, but the argument is what makes it act. 4. **ASK otherwise** → approval card. Deliberately *not* announced — the card already shows the call, and a second line would double it. 5. **NEVER** → returns an error *string* to the model, not an exception. The turn continues and the model sees the denial. +**"always" on an `ask_when` card grants `tool:argument`, not the tool.** That is +why step 2 checks grants before the bare name: a composite grant only lifts the +veto, so consulting the bare name first would make "always" a no-op that re-cards +forever on a tool nothing else approves. The grant is still narrow — it never +approves a call that doesn't carry that argument, and never covers a sibling +argument. + `ASK` is never terminal; it always resolves to NEVER or APPROVED. ### read_only is a narrow promise diff --git a/src/chief/gate_policy.py b/src/chief/gate_policy.py index d201e25a..3f6842c5 100644 --- a/src/chief/gate_policy.py +++ b/src/chief/gate_policy.py @@ -61,12 +61,26 @@ def decide( ) -> Decision: if tool_name in self.never: return Decision.NEVER - if self.pending_arguments(tool_name, arguments): - return Decision.ASK + if self._watched_present(tool_name, arguments): + # A call carrying a watched argument is decided by its grants + # alone — the bare name is neither required nor sufficient. This + # must come *before* the bare-name branch: a composite grant only + # lifts the veto, so falling through would leave "always" a no-op + # that re-cards forever on a tool nothing else approves. + if self.pending_arguments(tool_name, arguments): + return Decision.ASK + return Decision.APPROVED if "*" in self.approved or tool_name in self.approved: return Decision.APPROVED return Decision.ASK + def _watched_present( + self, tool_name: str, arguments: Mapping[str, Any] | None + ) -> tuple[str, ...]: + """The watched argument names this call actually carries.""" + watched = self.ask_when.get(tool_name) or () + return tuple(name for name in watched if name in (arguments or {})) + def pending_arguments( self, tool_name: str, arguments: Mapping[str, Any] | None = None ) -> tuple[str, ...]: @@ -74,10 +88,10 @@ def pending_arguments( Non-empty means the call must be carded no matter what the lists say. """ - watched = self.ask_when.get(tool_name) or () - present = (name for name in watched if name in (arguments or {})) return tuple( - name for name in present if grant_key(tool_name, name) not in self.approved + name + for name in self._watched_present(tool_name, arguments) + if grant_key(tool_name, name) not in self.approved ) def allow_always(self, tool_name: str) -> None: diff --git a/tests/test_gate.py b/tests/test_gate.py index 5f225d09..96afa049 100644 --- a/tests/test_gate.py +++ b/tests/test_gate.py @@ -94,6 +94,39 @@ def test_grant_covers_only_the_argument_it_names() -> None: assert policy.decide("peek", {"actions": [], "fill": "x"}) is Decision.ASK +def test_a_grant_approves_without_the_bare_name_being_approved() -> None: + """The tap that raised the card has to be enough on its own. + + A composite grant lifts the ask_when veto, but the decision used to fall + through to the bare-name check — so "always" on a tool that is not + independently approved granted nothing usable and re-carded forever. Every + other ask_when test seeds the bare name alongside the grant, which is why + this went unnoticed. + """ + policy = GatePolicy( + approved={grant_key("peek", "actions")}, + ask_when={"peek": ("actions",)}, + ) + assert policy.pending_arguments("peek", {"actions": []}) == () + assert policy.decide("peek", {"actions": []}) is Decision.APPROVED + # Still argument-scoped: a call not carrying a watched argument is decided + # by the ordinary lists, where `peek` is approved by nothing. + assert policy.decide("peek", {"url": "u"}) is Decision.ASK + + +def test_always_on_an_ask_when_card_stops_the_carding_next_time() -> None: + """The round trip the owner actually performs, at the policy layer. + + Mirrors what ``GatedTools._ask_card`` persists on an ALWAYS answer. + """ + policy = GatePolicy(ask_when={"peek": ("actions",)}) + call = {"actions": [{"click": "a"}]} + assert policy.decide("peek", call) is Decision.ASK + for argument in policy.pending_arguments("peek", call): + policy.allow_always(grant_key("peek", argument)) + assert policy.decide("peek", call) is Decision.APPROVED + + def make_gated( tmp_path: Path, answer: Approval,