diff --git a/grapharc/cli/agent.py b/grapharc/cli/agent.py index 17da119..ca67b82 100644 --- a/grapharc/cli/agent.py +++ b/grapharc/cli/agent.py @@ -39,6 +39,13 @@ DEFAULT_MAX_TOKENS = 100_000 DEFAULT_MAX_SECONDS = 300.0 +#: Sibling of the run's trace: the policy-document enforcement behind an +#: `agent --policy` run lands here as JSONL, via `grapharc.policy.audit` — +#: denials, and the decision-plus-outcome pair behind every approval request. +#: Allowances leave no record: the compiled policy grants them without +#: consulting the engine, so there is no document decision to write down. +POLICY_AUDIT_FILENAME = "policy-audit.jsonl" + def _accepts(fn: Any, param: str) -> bool: """Whether `fn` can be called with `param=`; unknown signatures say no.""" @@ -88,14 +95,25 @@ def build_registry(workspace: Path) -> tuple[Any, str]: ) -def build_policy(allow: list[str], deny: list[str], ask: list[str]) -> Any: - """Flags to a `PermissionPolicy`. Unmatched tools keep the policy's DENY default.""" +def build_policy( + allow: list[str], deny: list[str], ask: list[str], *, default: Any = None +) -> Any: + """Flags to a `PermissionPolicy`. + + Unmatched tools keep the `PermissionPolicy` DENY default — unless a + caller passes `default=ALLOW`, which is what the policy-document path + does: beside a document the flags are refinements, and an unmatched tool + is "no flag opinion", decided by the document alone. The default is `None` + meaning DENY rather than a `Decision` so this module keeps its lazy + `grapharc.harness` import. + """ from grapharc.harness import Decision, PermissionPolicy, PermissionRule rules = [PermissionRule(action=Decision.DENY, pattern=p) for p in deny] rules += [PermissionRule(action=Decision.ASK, pattern=p) for p in ask] rules += [PermissionRule(action=Decision.ALLOW, pattern=p) for p in allow] - return PermissionPolicy(rules=rules) + unmatched = Decision.DENY if default is None else default + return PermissionPolicy(rules=rules, default=unmatched) def _approval(as_json: bool, *, stream: Any = None) -> Any: @@ -117,6 +135,129 @@ def ask(tool_name: str, args: dict[str, Any]) -> bool: return ask +def _load_agent_document( + policy_path: Path, *, tenant: str | None, audit_path: Path +) -> tuple[Any, str]: + """Load and validate the document governing an agent run. + + Returns `(engine, tenant_name)`. Raises `PolicyError` — the shape a + missing or malformed file already fails with — for the two agent-specific + refusals as well: a document declaring no tool rules cannot govern tools, + and a tenant the document does not declare would deny every tool. Both + are refused before the audit log is built, so a refused run writes + nothing, not even an empty audit file. + """ + from grapharc.policy import AuditLog, PolicyEngine, PolicyError, load_document + from grapharc.policy.document import DEFAULT_TENANT, ResourceKind + + document = load_document(policy_path) + if not document.rules_for(ResourceKind.TOOL): + raise PolicyError( + f"policy document {policy_path} declares no tool rules, so it cannot " + "govern an agent run — it constrains nothing this command does" + ) + tenant_name = DEFAULT_TENANT if tenant is None else tenant + if not document.declares_tenant(tenant_name): + if tenant is None: + hint = "pass --tenant to name one" + else: + hint = "check the spelling of --tenant" + raise PolicyError( + f"tenant {tenant_name!r} is not declared by policy " + f"{document.version!r}; declared: {document.tenants!r} — {hint}" + ) + return PolicyEngine(document, audit=AuditLog(audit_path)), tenant_name + + +def combined_approval( + *, + engine: Any, + doc_policy: Any, + flag_policy: Any, + tenant: str, + as_json: bool, + stream: Any = None, + context: dict[str, Any] | None = None, +) -> Any: + """Approval honoring both a policy document and CLI `--ask` flags. + + Called by `Harness` when the combined policy answers ASK — so at least + one side asks and neither denies — and granted only when every asking + side grants: the document's approver role through `ApprovalRouter`, the + flags through the same terminal prompt `--ask` has always used. A role + handler that is missing, refuses, fails, or has nobody behind it (JSON + mode, redirected stdin) denies, exactly as the router already fails + closed; the role, rule and reason travel into the prompt so the human + approves a named thing, not a bare tool call. + """ + from grapharc.harness import Decision + from grapharc.policy.document import ResourceKind + + roles = sorted( + { + rule.approver_role + for rule in engine.document.rules_for(ResourceKind.TOOL) + if rule.effect is Decision.ASK and rule.approver_role + } + ) + + def handle_role(role: str) -> Any: + def handle(request: Any) -> bool: + # Same fail-closed guard `_approval` stands on: a prompt has no + # business in a pipe, and piped bytes are not consent. + source = stream or sys.stdin + if as_json or not getattr(source, "isatty", lambda: False)(): + return False + answer = input( + f"allow {request.subject} as {role} " + f"(rule {request.rule_id}: {request.reason})? [y/N] " + ).strip().lower() + return answer in ("y", "yes") + + return handle + + router = engine.approval_router( + {role: handle_role(role) for role in roles}, tenant=tenant + ) + flag_prompt = _approval(as_json, stream=stream) + + def approve(tool_name: str, args: dict[str, Any]) -> bool: + doc_asks = doc_policy.decide(tool_name) is Decision.ASK + flag_asks = flag_policy.decide(tool_name) is Decision.ASK + granted = True + if doc_asks: + decision = engine.check_tool(tool_name, tenant=tenant, context=context) + granted = router.route(decision, args=args).granted + if granted and flag_asks: + granted = flag_prompt(tool_name, args) + # Both asking sides granted; neither asking at all is a refusal, not + # an approval — unreachable from Harness, which only calls back on ASK. + return granted and (doc_asks or flag_asks) + + return approve + + +def document_denial_recorder( + *, engine: Any, doc_policy: Any, tenant: str, context: dict[str, Any] | None = None +) -> Any: + """An `on_denial` hook recording document-side denials, and only those. + + `Harness` fires the hook for every policy denial, including ones the CLI + flags caused. Recording a flag-side denial through the engine would write + ALLOW beside a refusal whenever the document permits the tool — a + contradiction in the audit log — so those stay trace-only, exactly as + before this command learned about documents. + """ + from grapharc.harness import Decision + + def record(tool_name: str) -> None: + if doc_policy.decide(tool_name) is not Decision.DENY: + return + engine.check_tool(tool_name, tenant=tenant, context=context) + + return record + + def run_agent( task: str, *, @@ -132,6 +273,8 @@ def run_agent( executor: str = "sandbox", system_prompt: str | None = None, run_id: str | None = None, + policy_path: Path | None = None, + tenant: str | None = None, as_json: bool = False, ) -> int: """Run one agent loop and report it. Returns the process exit code. @@ -140,10 +283,38 @@ def run_agent( an exhausted budget, an error — exits 1, because a script that ran an agent needs to know the task was not finished without parsing the reason first. + `policy_path` names a TOML policy document whose tool rules govern the run + beside the CLI flags: the document is the ceiling and the flags can only + narrow it — most restrictive wins, so a document DENY beats a flag allow + and a document ASK stays asked. Denials the document causes are recorded + to a policy audit file next to the trace; `--tenant` compiles the + document for one tenant and is refused without `--policy`, as `--policy` + itself is refused for delegated execution — `--executor claude-cli` or a + `claude-cli/*` model — which cannot enforce it. + `max_tokens=None` means the default ceiling on the governed path — and is the only value the delegated path accepts, because a ceiling it cannot enforce must be refused rather than silently unapplied. """ + if policy_path is not None and executor == "claude-cli": + # The delegated loop runs inside Claude Code, outside this process's + # policy, approval routing and audit — mapping a document onto CLI + # flags would claim an enforcement that is not there. Refused, like + # the token ceiling the same path cannot honor below. + return fail( + "--policy cannot be enforced under --executor claude-cli: the delegated " + "loop runs outside this process's policy and audit. Drop --executor " + "claude-cli for a governed run", + as_json=as_json, + command="agent", + ) + if tenant is not None and policy_path is None: + return fail( + "--tenant names whose rules a policy document enforces, so it needs " + "--policy to mean anything", + as_json=as_json, + command="agent", + ) if executor == "claude-cli": # The whole loop is Claude Code's; nothing below (registry, harness, # gateway model) applies. `--model` semantics shift too: the delegated @@ -184,7 +355,14 @@ def run_agent( from grapharc.runtime.budget import Budget, BudgetExceeded, BudgetMeter, deadline_guard from grapharc.runtime.graph import RunContext - allow = allow or ["*"] + if policy_path is None: + allow = allow or ["*"] + else: + # No implicit allow-all beside a document: the document's default + # governs unmatched tools, and an implicit `*` would permit what a + # default-deny document refuses. Explicit --allow still votes allow — + # it just cannot outvote the document (see CombinedPolicy). + allow = allow or [] deny = deny or [] ask = ask or [] workspace = Path(workspace).expanduser().resolve() @@ -202,15 +380,74 @@ def run_agent( except optional.Unavailable as exc: return fail(str(exc), as_json=as_json, command="agent") - policy = build_policy(allow, deny, ask) + governed: tuple[Any, str, Path] | None = None + if policy_path is not None: + from grapharc.policy import PolicyError + + try: + audit_path = trace_path.parent / POLICY_AUDIT_FILENAME + engine, tenant_name = _load_agent_document( + Path(policy_path), + tenant=tenant, + audit_path=audit_path, + ) + except PolicyError as exc: + return fail(str(exc), as_json=as_json, command="agent", task=task) + governed = (engine, tenant_name, audit_path) + + if governed is None: + policy = build_policy(allow, deny, ask) + approval: Any = _approval(as_json) + on_denial: Any = None + else: + from grapharc.harness import CombinedPolicy, Decision + + engine, tenant_name, _ = governed + doc_policy = engine.permission_policy(tenant=tenant_name) + # Flags beside a document vote no opinion by default (ALLOW): the + # document decides unmatched tools, and --deny/--ask narrow it. + flag_policy = build_policy(allow, deny, ask, default=Decision.ALLOW) + policy = CombinedPolicy(policies=[doc_policy, flag_policy]) + audit_context = {"command": "agent", "run_id": run_id} + approval = combined_approval( + engine=engine, + doc_policy=doc_policy, + flag_policy=flag_policy, + tenant=tenant_name, + as_json=as_json, + context=audit_context, + ) + on_denial = document_denial_recorder( + engine=engine, doc_policy=doc_policy, tenant=tenant_name, context=audit_context + ) harness = Harness( registry, policy, executor=LocalExecutor() if executor == "local" else None, workspace=str(workspace), - approval=_approval(as_json), + approval=approval, + on_denial=on_denial, ) visible = [spec.name for spec in harness.visible_tools()] + # Present only when a document governed the run: the no-document payload + # keeps exactly the keys it has always had. + governance_extra: dict[str, Any] = {} + if governed is not None: + from grapharc.policy.document import ResourceKind + + _engine, _tenant_name, _audit_path = governed + _document = _engine.document + governance_extra = { + "policy_document": { + "source": "flag", + "path": str(policy_path), + "version": _document.version, + "digest": _engine.digest, + "tenant": _tenant_name, + "tool_rules": len(_document.rules_for(ResourceKind.TOOL)), + }, + "policy_audit": str(_audit_path), + } try: from grapharc.gateway import get_model @@ -221,6 +458,22 @@ def run_agent( f"could not build model {model_spec!r}: {exc}", as_json=as_json, command="agent" ) + if governed is not None: + from grapharc.harness.agent import is_claude_cli + + # Delegation triggers on the model, not on --executor: a claude-cli + # model would hand the loop to a subprocess outside the harness this + # policy was just compiled for. Same predicate AgentNode delegates on, + # so the two cannot disagree. + if is_claude_cli(model): + return fail( + "--policy cannot be enforced for a claude-cli model: the loop " + "runs inside Claude Code, outside this process's policy and " + "audit. Use a tool-calling backend for a governed run", + as_json=as_json, + command="agent", + ) + trace = TraceRecorder(trace_path) # The loop's own turn cap bounds iterations, so the meter is left to bound # the two things it alone can see: spend and wall clock. Setting both would @@ -255,6 +508,7 @@ def run_agent( "tools_from": f"grapharc.tools.{entry_point}", "policy": {"allow": allow, "ask": ask, "deny": deny}, "tools_visible": visible, + **governance_extra, } try: @@ -303,6 +557,16 @@ def count(number: int) -> str: """ return style.err(str(number)) if number else str(number) + policy_value = ( + f"{style.dim('allow=')}{allow} {style.dim('ask=')}{ask} {style.dim('deny=')}{deny}" + ) + if governed is not None: + _, _tenant_name, _audit_path = governed + policy_value += ( + f" {style.dim('document=')}{policy_path}" + f" {style.dim('tenant=')}{_tenant_name}" + f" {style.dim('audit=')}{_audit_path}" + ) lines = [ style.kv("task", task, width=width), style.kv("model", model_spec, width=width, tint=style.accent), @@ -314,7 +578,7 @@ def count(number: int) -> str: ), style.kv( "policy", - f"{style.dim('allow=')}{allow} {style.dim('ask=')}{ask} {style.dim('deny=')}{deny}", + policy_value, width=width, ), "", @@ -357,5 +621,7 @@ def count(number: int) -> str: "CORE_TOOL_ENTRY_POINTS", "build_policy", "build_registry", + "combined_approval", + "document_denial_recorder", "run_agent", ] diff --git a/grapharc/cli/main.py b/grapharc/cli/main.py index 0640b72..6a8280a 100644 --- a/grapharc/cli/main.py +++ b/grapharc/cli/main.py @@ -566,6 +566,8 @@ def _cmd_agent(args: argparse.Namespace) -> int: max_seconds=args.max_seconds, executor=args.executor, system_prompt=args.system_prompt, + policy_path=args.policy, + tenant=args.tenant, run_id=args.run_id, as_json=args.json, ) @@ -1130,6 +1132,24 @@ def build_parser() -> argparse.ArgumentParser: "subscription (default: sandbox)" ), ) + agent.add_argument( + "--policy", + type=Path, + default=None, + metavar="PATH", + help=( + "TOML policy document governing this agent's tools. The document " + "is the ceiling: --deny/--ask narrow it further, --allow cannot " + "widen it, and denials are recorded to policy-audit.jsonl next to " + "the trace" + ), + ) + agent.add_argument( + "--tenant", + default=None, + metavar="NAME", + help="tenant to compile --policy for (needs --policy)", + ) agent.add_argument("--system-prompt", default=None) agent.set_defaults(handler=_cmd_agent) diff --git a/grapharc/harness/__init__.py b/grapharc/harness/__init__.py index 3cc6566..1549327 100644 --- a/grapharc/harness/__init__.py +++ b/grapharc/harness/__init__.py @@ -8,10 +8,11 @@ tool_schema, tool_schemas, ) -from grapharc.harness.core import ApprovalCallback, Harness +from grapharc.harness.core import ApprovalCallback, DenialCallback, Harness from grapharc.harness.executor import LocalExecutor, SandboxedExecutor, SandboxViolation from grapharc.harness.hooks import HookAction, HookDecision, PostHook, PreHook from grapharc.harness.permissions import ( + CombinedPolicy, Decision, PermissionDenied, PermissionPolicy, @@ -25,6 +26,8 @@ "AgentNode", "AgentResult", "ApprovalCallback", + "DenialCallback", + "CombinedPolicy", "Decision", "Harness", "HookAction", diff --git a/grapharc/harness/agent.py b/grapharc/harness/agent.py index 8cd58db..a24338c 100644 --- a/grapharc/harness/agent.py +++ b/grapharc/harness/agent.py @@ -355,8 +355,12 @@ class DelegatedToolUseWarning(UserWarning): ) -def _is_claude_cli(model: Any) -> bool: - """Is this the Claude CLI backend? +def is_claude_cli(model: Any) -> bool: + """Is this the Claude CLI backend — the loop that delegates? + + `AgentNode` delegates on this predicate, and `grapharc agent --policy` + refuses on it; one shared test so the two can never disagree about what + "delegated" means. Matched on `_llm_type` rather than `isinstance`, so this module does not import the gateway, and rather than "does it lack bind_tools" — which is @@ -419,7 +423,7 @@ def __init__( #: True when the backend is the Claude CLI, which has no tool-calling #: wire format and therefore cannot be driven as a raw model. The loop #: is handed to Claude Code instead — see `_run_delegated`. - self.delegated = _is_claude_cli(model) + self.delegated = is_claude_cli(model) self.delegated_mode = delegated_mode if self.delegated: if delegated_mode == "bypass": diff --git a/grapharc/harness/core.py b/grapharc/harness/core.py index be3266a..80a075c 100644 --- a/grapharc/harness/core.py +++ b/grapharc/harness/core.py @@ -9,6 +9,8 @@ from __future__ import annotations +import logging +import threading from collections.abc import Callable from typing import Any @@ -17,8 +19,14 @@ from grapharc.harness.permissions import Decision, PermissionDenied, PermissionPolicy from grapharc.harness.tools import ToolRegistry, ToolSpec +_log = logging.getLogger(__name__) + # (tool_name, args) -> approved? Human checkpoints implement this. ApprovalCallback = Callable[[str, dict[str, Any]], bool] +# (tool_name) -> None. Fired when the policy denies a call, so a caller that +# records enforcement elsewhere (a policy document's audit log) hears about +# the denials this object otherwise answers silently. +DenialCallback = Callable[[str], None] class Harness: @@ -31,6 +39,7 @@ def __init__( pre_hooks: tuple[PreHook, ...] = (), post_hooks: tuple[PostHook, ...] = (), approval: ApprovalCallback | None = None, + on_denial: DenialCallback | None = None, workspace: str | None = None, ) -> None: self.registry = registry @@ -50,6 +59,11 @@ def __init__( self.pre_hooks = pre_hooks self.post_hooks = post_hooks self.approval = approval + self.on_denial = on_denial + #: Denial-hook exceptions swallowed so far. Non-zero means enforcement + #: records are missing somewhere downstream — see `_notify_denial`. + self.denial_hook_errors = 0 + self._denial_lock = threading.Lock() def visible_tools(self) -> list[ToolSpec]: """The tool schemas a model may see — policy-filtered before exposure.""" @@ -62,6 +76,7 @@ def call(self, tool_name: str, args: dict[str, Any]) -> Any: decision = self.policy.decide(tool_name) if decision is Decision.DENY: + self._notify_denial(tool_name) raise PermissionDenied(f"tool {tool_name!r} denied by policy") if decision is Decision.ASK: # Fail closed: no approval channel means no approval. @@ -87,3 +102,25 @@ def call(self, tool_name: str, args: dict[str, Any]) -> Any: for post in self.post_hooks: result = post(tool_name, dict(args), result) return result + + def _notify_denial(self, tool_name: str) -> None: + """Tell `on_denial` about a refused call, without fail. + + The hook runs before the `PermissionDenied` it annotates, and its + exceptions are counted in `denial_hook_errors` and swallowed: a + recorder must never turn a denied call into a tool error, which is + what an exception here would become one frame up in `AgentNode`. + Only the policy-DENY branch notifies — an approval refused by a + human is the approval path's record to write, and an unknown tool + is not a policy decision at all. + """ + if self.on_denial is None: + return + try: + self.on_denial(tool_name) + except Exception: + with self._denial_lock: + self.denial_hook_errors += 1 + _log.exception( + "denial hook raised for tool %r; the denial stands", tool_name + ) diff --git a/grapharc/harness/permissions.py b/grapharc/harness/permissions.py index 5b082c7..1e0e8a9 100644 --- a/grapharc/harness/permissions.py +++ b/grapharc/harness/permissions.py @@ -18,7 +18,7 @@ from enum import StrEnum from fnmatch import fnmatch -from pydantic import BaseModel +from pydantic import BaseModel, field_validator, model_validator class Decision(StrEnum): @@ -90,3 +90,61 @@ def decide(self, tool_name: str) -> Decision: if rule.action == tier and rule.matches(tool_name): return tier return self.default + + +class CombinedPolicy(PermissionPolicy): + """Two or more policies consulted together; the most restrictive wins. + + Tiered semantics lifted from rules to policies: a tool is DENY when any + side denies it, ASK when any side asks and none denies, and ALLOW only + when every side allows. The order of `policies` does not matter — unlike + rule order within one policy there is no first match here, only the + strictest verdict. + + The use case is a ceiling plus refinements: a policy document compiled by + `PolicyEngine.permission_policy()` sets the maximum authority, and a + flag-built policy can only narrow it. A CLI `--allow` votes ALLOW exactly + as before, but it can never outvote a document DENY or quietly demote a + document ASK — most-restrictive is precisely "flags cannot widen the + document". The flag side therefore carries default ALLOW: with no flag + opinion on a tool, the document decides it alone. + + `rules` and `default` are inherited and rejected, not merged: this answers + only from `policies`, and a rule written on the combination itself would + be silently ignored, which for a DENY rule fails open. `Harness` and + `ToolRegistry.visible` speak `PermissionPolicy`, so this subclasses it and + drops into either unchanged. + """ + + policies: list[PermissionPolicy] = [] + + @field_validator("policies") + @classmethod + def _require_policies( + cls, policies: list[PermissionPolicy] + ) -> list[PermissionPolicy]: + if not policies: + raise ValueError("CombinedPolicy needs at least one policy to combine") + return policies + + @model_validator(mode="after") + def _forbid_own_rules(self) -> CombinedPolicy: + if self.rules or self.default is not Decision.DENY: + raise ValueError( + "CombinedPolicy answers from `policies`, not from its own rules: " + "append another PermissionPolicy instead of writing rules here" + ) + return self + + def decide(self, tool_name: str) -> Decision: + if not self.policies: + # Unreachable from the constructor — the validator refuses an + # empty list — but reachable by post-construction mutation, and + # an empty verdict set must deny, never fall through to allow. + return Decision.DENY + decisions = {policy.decide(tool_name) for policy in self.policies} + if Decision.DENY in decisions: + return Decision.DENY + if Decision.ASK in decisions: + return Decision.ASK + return Decision.ALLOW diff --git a/tests/test_cli.py b/tests/test_cli.py index c997320..0e8f00a 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -23,6 +23,7 @@ import subprocess import sys import time +from io import StringIO from pathlib import Path from types import ModuleType @@ -30,11 +31,17 @@ from langchain_core.messages import AIMessage from pydantic import BaseModel -from grapharc.cli.agent import _approval +from grapharc.cli.agent import ( + _approval, + build_policy, + combined_approval, + document_denial_recorder, +) from grapharc.cli.main import build_parser, main from grapharc.examples.stage0_dag import DEMO_DOC, build_stage0 -from grapharc.harness import ToolSpec +from grapharc.harness import Decision, ToolSpec from grapharc.observe.trace import TraceRecorder +from grapharc.policy import PolicyEngine # -- harness ------------------------------------------------------------------ @@ -2481,3 +2488,913 @@ def test_default_flag_forces_the_builtin_kinds(tmp_path, monkeypatch, capsys): payload = json.loads(capsys.readouterr().out) assert code == 0 assert payload["registry"] == "grapharc.stdlib:build_registry" + + +# -- agent --policy (issue #6) -------------------------------------------------- + + +DOC_DENY_SHELL = """ +version = "1.0.0" +name = "agent-cli-test" +default = "deny" + +[[rule]] +id = "no-shell" +resource = "tool" +match = "run_command" +effect = "deny" +reason = "the shell is never permitted" + +[[rule]] +id = "notes-ok" +resource = "tool" +match = "write_note" +effect = "allow" +""" + +DOC_ALLOW_WRITES = """ +version = "1.0.0" +name = "agent-cli-test" +default = "deny" + +[[rule]] +id = "notes-ok" +resource = "tool" +match = "write_*" +effect = "allow" +""" + +DOC_ASK_WRITES = """ +version = "1.0.0" +name = "agent-cli-test" +default = "deny" + +[[rule]] +id = "notes-ask" +resource = "tool" +match = "write_note" +effect = "ask" +approver_role = "reviewer" +reason = "writes need a human" +""" + +DOC_TENANTED = """ +version = "1.0.0" +name = "agent-cli-test" +default = "deny" +tenants = ["acme", "globex"] + +[[rule]] +id = "acme-notes" +resource = "tool" +match = "write_*" +effect = "allow" +tenant = "acme" +""" + +DOC_DENY_WRITES = """ +version = "1.0.0" +name = "agent-cli-test" +default = "deny" + +[[rule]] +id = "no-writes" +resource = "tool" +match = "write_note" +effect = "deny" +reason = "read-only run" +""" + +DOC_NO_TOOLS = """ +version = "1.0.0" +name = "agent-cli-test" +default = "deny" + +[[rule]] +id = "chain" +resource = "edge" +match = "triage->fix" +effect = "allow" +""" + + +def _write_doc(path: Path, text: str) -> Path: + path.write_text(text.strip() + "\n", encoding="utf-8") + return path + + +def _read_jsonl(path) -> list[dict]: + """Audit/trace lines as dicts. A run that recorded nothing has no file yet — + the same contract TraceRecorder keeps — which reads as zero records.""" + target = Path(path) + if not target.exists(): + return [] + return [ + json.loads(line) + for line in target.read_text(encoding="utf-8").splitlines() + if line.strip() + ] + + +class _TtyInput(StringIO): + """Piped bytes with a tty face, for the interactive approval path.""" + + def isatty(self) -> bool: + return True + + +def _refuse_to_prompt(*args, **kwargs): + raise AssertionError("prompted for approval in a non-interactive run") + + +def test_agent_policy_document_denies_a_tool_end_to_end( + tmp_path, capsys, scripted_model, stub_tools +): + """Issue #6: the document governs the run — denied tools are undescribed, + unexecuted, and recorded in both the audit log and the trace.""" + stub_tools() + workspace = tmp_path / "ws" + policy = _write_doc(tmp_path / "policy.toml", DOC_DENY_SHELL) + model = scripted_model( + [ + {"tools": [("run_command", {"argv": "id"})]}, + {"tools": [("write_note", {"path": "note.txt", "content": "hello"})]}, + {"content": "wrote the note"}, + ] + ) + code, payload, _ = call_json( + [ + "agent", + "do the thing", + "--model", + "mock/x", + "--workspace", + str(workspace), + "--policy", + str(policy), + "--run-id", + "doc-test", + # Local execution: the sandbox executor forks, which Windows + # cannot do — and confinement is orthogonal to what this test + # proves (policy visibility, denial, audit, trace). + "--executor", + "local", + ], + capsys, + ) + assert code == 0 + # Selectively governed: the denied tool is undescribed, the allowed one runs. + assert model.bound_tools == ["write_note"] + assert payload["tools_visible"] == ["write_note"] + assert payload["policy"] == {"allow": [], "ask": [], "deny": []} + assert (workspace / "note.txt").read_text(encoding="utf-8") == "hello" + assert payload["tool_calls"][0]["tool"] == "run_command" + assert payload["tool_calls"][0]["status"] == "denied" + assert payload["tool_calls"][0]["refused_by"] == "policy" + assert payload["tool_calls"][1]["status"] == "ok" + + # The denial is attributable: rule, version, digest, tenant, effect. + audit = _read_jsonl(payload["policy_audit"]) + decision = next( + e for e in audit if e["kind"] == "decision" and e["subject"] == "run_command" + ) + assert decision["effect"] == "deny" + assert decision["rule_id"] == "no-shell" + assert decision["policy_version"] == "1.0.0" + assert decision["tenant"] == "default" + assert decision["policy_digest"] == payload["policy_document"]["digest"] + + # ...and the same refusal is on the run trace, not only in the audit log. + trace = _read_jsonl(payload["trace"]) + refusal = next( + e + for e in trace + if e.get("phase") == "tool" + and (e.get("state_delta") or {}).get("tool") == "run_command" + ) + assert refusal["state_delta"]["status"] == "denied" + assert refusal["state_delta"]["refused_by"] == "policy" + assert "PERMISSION_DENIED" in (refusal.get("error") or "") + + +def test_agent_flag_allow_cannot_widen_a_document_deny( + tmp_path, capsys, scripted_model, stub_tools +): + stub_tools() + workspace = tmp_path / "ws" + policy = _write_doc(tmp_path / "policy.toml", DOC_DENY_SHELL) + model = scripted_model( + [ + {"tools": [("run_command", {"argv": "id"})]}, + {"content": "blocked"}, + ] + ) + code, payload, _ = call_json( + [ + "agent", + "t", + "--model", + "mock/x", + "--workspace", + str(workspace), + "--policy", + str(policy), + "--allow", + "run_command", + ], + capsys, + ) + assert code == 0 + assert payload["policy"]["allow"] == ["run_command"] + assert "run_command" not in model.bound_tools + assert payload["tool_calls"][0]["status"] == "denied" + audit = _read_jsonl(payload["policy_audit"]) + assert any( + e["kind"] == "decision" + and e["subject"] == "run_command" + and e["effect"] == "deny" + and e["rule_id"] == "no-shell" + for e in audit + ) + + +def test_agent_flag_deny_narrows_a_document_allow( + tmp_path, capsys, scripted_model, stub_tools +): + stub_tools() + workspace = tmp_path / "ws" + policy = _write_doc(tmp_path / "policy.toml", DOC_ALLOW_WRITES) + model = scripted_model( + [ + {"tools": [("write_note", {"path": "note.txt", "content": "hello"})]}, + {"content": "blocked"}, + ] + ) + code, payload, _ = call_json( + [ + "agent", + "t", + "--model", + "mock/x", + "--workspace", + str(workspace), + "--policy", + str(policy), + "--deny", + "write_note", + ], + capsys, + ) + assert code == 0 + assert "write_note" not in (model.bound_tools or []) + assert payload["tool_calls"][0]["status"] == "denied" + assert not (workspace / "note.txt").exists() + # The flag caused this refusal, not the document — recording it through + # the engine would write ALLOW beside a denial, so the audit stays empty. + assert _read_jsonl(payload["policy_audit"]) == [] + + +def test_agent_flag_allow_reaffirms_a_document_allow( + tmp_path, capsys, scripted_model, stub_tools +): + stub_tools() + workspace = tmp_path / "ws" + policy = _write_doc(tmp_path / "policy.toml", DOC_ALLOW_WRITES) + scripted_model( + [ + {"tools": [("write_note", {"path": "note.txt", "content": "hello"})]}, + {"content": "wrote the note"}, + ] + ) + code, payload, _ = call_json( + [ + "agent", + "t", + "--model", + "mock/x", + "--workspace", + str(workspace), + "--policy", + str(policy), + "--allow", + "write_note", + # Local execution: the sandbox executor forks, which Windows + # cannot do — and confinement is orthogonal to the reaffirmed + # allow this test proves. + "--executor", + "local", + ], + capsys, + ) + assert code == 0 + assert payload["tool_calls"][0]["status"] == "ok" + assert (workspace / "note.txt").read_text(encoding="utf-8") == "hello" + + +def test_agent_flag_allow_cannot_bypass_a_document_ask( + tmp_path, capsys, scripted_model, stub_tools, monkeypatch +): + """Document ASK plus flag allow stays asked — and in JSON mode the human + it asks for does not exist, so the call is refused, not granted.""" + monkeypatch.setattr("builtins.input", _refuse_to_prompt) + stub_tools() + workspace = tmp_path / "ws" + policy = _write_doc(tmp_path / "policy.toml", DOC_ASK_WRITES) + model = scripted_model( + [ + {"tools": [("write_note", {"path": "note.txt", "content": "hello"})]}, + {"content": "blocked"}, + ] + ) + code, payload, _ = call_json( + [ + "agent", + "t", + "--model", + "mock/x", + "--workspace", + str(workspace), + "--policy", + str(policy), + "--allow", + "write_note", + ], + capsys, + ) + assert code == 0 + # Asked tools stay visible — ask gates the call, it does not hide the schema. + assert "write_note" in model.bound_tools + assert payload["tool_calls"][0]["status"] == "denied" + assert not (workspace / "note.txt").exists() + audit = _read_jsonl(payload["policy_audit"]) + decision = next(e for e in audit if e["kind"] == "decision") + assert decision["effect"] == "ask" + assert decision["rule_id"] == "notes-ask" + assert decision["approver_role"] == "reviewer" + approval = next(e for e in audit if e["kind"] == "approval") + assert approval["granted"] is False + assert approval["approver_role"] == "reviewer" + # The guard refused — had `input()` been reached, the router would have + # recorded the stub's AssertionError as a handler failure instead. + assert approval["reason"] == "reviewer refused approval" + + +def test_agent_document_deny_plus_flag_deny_is_still_denied( + tmp_path, capsys, scripted_model, stub_tools +): + stub_tools() + workspace = tmp_path / "ws" + policy = _write_doc(tmp_path / "policy.toml", DOC_DENY_SHELL) + scripted_model( + [ + {"tools": [("run_command", {"argv": "id"})]}, + {"content": "blocked"}, + ] + ) + code, payload, _ = call_json( + [ + "agent", + "t", + "--model", + "mock/x", + "--workspace", + str(workspace), + "--policy", + str(policy), + "--deny", + "run_command", + ], + capsys, + ) + assert code == 0 + assert payload["tool_calls"][0]["status"] == "denied" + audit = _read_jsonl(payload["policy_audit"]) + assert any(e["kind"] == "decision" and e["rule_id"] == "no-shell" for e in audit) + + +def test_agent_denied_tool_never_reaches_the_executor( + tmp_path, capsys, scripted_model, stub_tools +): + stub_tools() + workspace = tmp_path / "ws" + policy = _write_doc(tmp_path / "policy.toml", DOC_DENY_WRITES) + model = scripted_model( + [ + {"tools": [("write_note", {"path": "note.txt", "content": "hello"})]}, + {"content": "blocked"}, + ] + ) + code, payload, _ = call_json( + [ + "agent", + "t", + "--model", + "mock/x", + "--workspace", + str(workspace), + "--policy", + str(policy), + ], + capsys, + ) + assert code == 0 + assert "write_note" not in (model.bound_tools or []) + assert payload["tool_calls"][0]["status"] == "denied" + assert payload["tool_calls"][0]["refused_by"] == "policy" + assert not (workspace / "note.txt").exists() + + +def test_agent_policy_tenant_allow_and_deny( + tmp_path, capsys, scripted_model, stub_tools +): + """One tenant's grant is not another's; the ungranted tenant is denied.""" + stub_tools() + policy = _write_doc(tmp_path / "policy.toml", DOC_TENANTED) + + granted_ws = tmp_path / "granted" + scripted_model( + [ + {"tools": [("write_note", {"path": "note.txt", "content": "hello"})]}, + {"content": "wrote the note"}, + ] + ) + code, granted, _ = call_json( + [ + "agent", + "t", + "--model", + "mock/x", + "--workspace", + str(granted_ws), + "--policy", + str(policy), + "--tenant", + "acme", + # Local execution on both halves of this comparison: the + # sandbox executor forks, which Windows cannot do. + "--executor", + "local", + ], + capsys, + ) + assert code == 0 + assert granted["tool_calls"][0]["status"] == "ok" + assert (granted_ws / "note.txt").read_text(encoding="utf-8") == "hello" + assert granted["policy_document"]["tenant"] == "acme" + + refused_ws = tmp_path / "refused" + scripted_model( + [ + {"tools": [("write_note", {"path": "note.txt", "content": "hello"})]}, + {"content": "blocked"}, + ] + ) + code, refused, _ = call_json( + [ + "agent", + "t", + "--model", + "mock/x", + "--workspace", + str(refused_ws), + "--policy", + str(policy), + "--tenant", + "globex", + "--executor", + "local", + ], + capsys, + ) + assert code == 0 + assert refused["tool_calls"][0]["status"] == "denied" + assert not (refused_ws / "note.txt").exists() + audit = _read_jsonl(refused["policy_audit"]) + decision = next(e for e in audit if e["kind"] == "decision") + assert decision["effect"] == "deny" + assert decision["tenant"] == "globex" + assert decision["rule_id"] is None # the document default, not a rule + + +def test_agent_policy_unknown_tenant_is_refused_upfront(tmp_path, capsys): + policy = _write_doc(tmp_path / "policy.toml", DOC_TENANTED) + code, payload, _ = call_json( + [ + "agent", + "t", + "--model", + "mock/x", + "--workspace", + str(tmp_path / "ws"), + "--policy", + str(policy), + "--tenant", + "nope", + ], + capsys, + ) + assert code == 2 + assert payload["ok"] is False + assert "nope" in payload["error"] + assert "declared" in payload["error"] + + +def test_agent_policy_missing_tenant_is_refused_when_tenants_are_listed( + tmp_path, capsys +): + policy = _write_doc(tmp_path / "policy.toml", DOC_TENANTED) + code, payload, _ = call_json( + [ + "agent", + "t", + "--model", + "mock/x", + "--workspace", + str(tmp_path / "ws"), + "--policy", + str(policy), + ], + capsys, + ) + assert code == 2 + assert "'default'" in payload["error"] + assert "pass --tenant" in payload["error"] + + +def test_agent_policy_ask_fails_closed_in_json_mode( + tmp_path, capsys, scripted_model, stub_tools, monkeypatch +): + """JSON means no human is reading: ask denies, never prompts, and the + refusal is structured — the run still completes so the document's shape + is visible instead of a traceback.""" + monkeypatch.setattr("builtins.input", _refuse_to_prompt) + stub_tools() + workspace = tmp_path / "ws" + policy = _write_doc(tmp_path / "policy.toml", DOC_ASK_WRITES) + scripted_model( + [ + {"tools": [("write_note", {"path": "note.txt", "content": "hello"})]}, + {"content": "blocked"}, + ] + ) + code, payload, _ = call_json( + [ + "agent", + "t", + "--model", + "mock/x", + "--workspace", + str(workspace), + "--policy", + str(policy), + ], + capsys, + ) + assert code == 0 + assert payload["denied"] == 1 + assert payload["tool_calls"][0]["refused_by"] == "policy" + assert not (workspace / "note.txt").exists() + audit = _read_jsonl(payload["policy_audit"]) + assert any(e["kind"] == "decision" and e["effect"] == "ask" for e in audit) + approval = next(e for e in audit if e["kind"] == "approval") + assert approval["granted"] is False + assert approval["reason"] == "reviewer refused approval" + + +def test_agent_policy_ask_grants_interactively( + tmp_path, capsys, scripted_model, stub_tools, monkeypatch +): + """On a terminal the role, rule and reason reach the human, and a yes runs.""" + stub_tools() + workspace = tmp_path / "ws" + policy = _write_doc(tmp_path / "policy.toml", DOC_ASK_WRITES) + scripted_model( + [ + {"tools": [("write_note", {"path": "note.txt", "content": "hello"})]}, + {"content": "wrote the note"}, + ] + ) + monkeypatch.setattr("sys.stdin", _TtyInput("y\n")) + code, out, _ = call( + [ + "agent", + "t", + "--model", + "mock/x", + "--workspace", + str(workspace), + "--policy", + str(policy), + # Local execution: the sandbox executor forks, which Windows + # cannot do — and confinement is orthogonal to the granted + # approval this test proves. + "--executor", + "local", + ], + capsys, + ) + assert code == 0 + assert (workspace / "note.txt").read_text(encoding="utf-8") == "hello" + assert "reviewer" in out + assert "notes-ask" in out + + +def test_agent_policy_refuses_the_delegated_executor(tmp_path, capsys): + """A document mapped onto Claude Code flags would claim an enforcement + that is not there — refused before the (missing) file is even read.""" + code, payload, _ = call_json( + [ + "agent", + "t", + "--executor", + "claude-cli", + "--workspace", + str(tmp_path / "ws"), + "--policy", + str(tmp_path / "nope.toml"), + ], + capsys, + ) + assert code == 2 + assert "--policy" in payload["error"] + assert "claude-cli" in payload["error"] + + +def test_agent_tenant_without_policy_is_refused(tmp_path, capsys): + code, payload, _ = call_json( + ["agent", "t", "--workspace", str(tmp_path / "ws"), "--tenant", "acme"], + capsys, + ) + assert code == 2 + assert "--tenant" in payload["error"] + assert "--policy" in payload["error"] + + +def test_agent_policy_missing_file_is_a_structured_error(tmp_path, capsys): + code, payload, _ = call_json( + [ + "agent", + "t", + "--model", + "mock/x", + "--workspace", + str(tmp_path / "ws"), + "--policy", + str(tmp_path / "nope.toml"), + ], + capsys, + ) + assert code == 2 + assert payload["ok"] is False + assert payload["command"] == "agent" + assert "nope.toml" in payload["error"] + + +def test_agent_policy_without_tool_rules_is_refused(tmp_path, capsys): + """A planner-only document constrains nothing an agent does — running + under it would report a policy that is not in force.""" + policy = _write_doc(tmp_path / "policy.toml", DOC_NO_TOOLS) + code, payload, _ = call_json( + [ + "agent", + "t", + "--model", + "mock/x", + "--workspace", + str(tmp_path / "ws"), + "--policy", + str(policy), + ], + capsys, + ) + assert code == 2 + assert "tool rules" in payload["error"] + + +def test_agent_policy_provenance_is_reported( + tmp_path, capsys, scripted_model, stub_tools +): + """The payload says which document governed the run — and the digest it + names is the one the audit records carry.""" + stub_tools() + workspace = tmp_path / "ws" + policy = _write_doc(tmp_path / "policy.toml", DOC_DENY_SHELL) + scripted_model( + [ + {"tools": [("run_command", {"argv": "id"})]}, + {"tools": [("write_note", {"path": "note.txt", "content": "hello"})]}, + {"content": "done"}, + ] + ) + code, payload, _ = call_json( + [ + "agent", + "t", + "--model", + "mock/x", + "--workspace", + str(workspace), + "--policy", + str(policy), + # Local execution: the sandbox executor forks, which Windows + # cannot do — and confinement is orthogonal to the provenance + # this test proves. + "--executor", + "local", + ], + capsys, + ) + assert code == 0 + document = payload["policy_document"] + assert document["source"] == "flag" + assert document["path"] == str(policy) + assert document["version"] == "1.0.0" + assert document["tenant"] == "default" + assert document["tool_rules"] == 2 + assert len(document["digest"]) == 64 + audit = _read_jsonl(payload["policy_audit"]) + assert audit, "a governed run that denied a tool must have audit records" + assert {e["policy_digest"] for e in audit} == {document["digest"]} + + text_ws = tmp_path / "text-ws" + scripted_model([{"content": "nothing to do"}]) + code, out, _ = call( + [ + "agent", + "t", + "--model", + "mock/x", + "--workspace", + str(text_ws), + "--policy", + str(policy), + ], + capsys, + ) + assert code == 0 + assert "document=" in out + assert "tenant=default" in out + assert "audit=" in out + + +def test_agent_combined_approval_grants_only_when_every_asking_side_grants(monkeypatch): + engine = PolicyEngine.from_toml(DOC_ASK_WRITES) + doc_policy = engine.permission_policy(tenant="default") + flag_open = build_policy(["*"], [], []) + flag_ask = build_policy([], [], ["write_note"]) + + # Answers travel through `builtins.input`, the way the handlers read + # them: `stream` only decides whether a terminal is behind stdin. + prompts: list[str] = [] + script: list[str] = [] + + def respond(prompt: str) -> str: + prompts.append(prompt) + return script.pop(0) + + monkeypatch.setattr("builtins.input", respond) + + approve = combined_approval( + engine=engine, + doc_policy=doc_policy, + flag_policy=flag_open, + tenant="default", + as_json=False, + stream=_TtyInput(), + ) + script.append("y") + assert approve("write_note", {"path": "n"}) is True + assert script == [] # the answer was consumed, exactly once + assert len(prompts) == 1 + assert "reviewer" in prompts[0] + assert "notes-ask" in prompts[0] + + both = combined_approval( + engine=engine, + doc_policy=doc_policy, + flag_policy=flag_ask, + tenant="default", + as_json=False, + stream=_TtyInput(), + ) + script.extend(["y", "y"]) + assert both("write_note", {}) is True + assert script == [] + + vetoed = combined_approval( + engine=engine, + doc_policy=doc_policy, + flag_policy=flag_ask, + tenant="default", + as_json=False, + stream=_TtyInput(), + ) + script.extend(["y", "n"]) + assert vetoed("write_note", {}) is False + assert script == [] + + asked = len(prompts) + refused = combined_approval( + engine=engine, + doc_policy=doc_policy, + flag_policy=flag_open, + tenant="default", + as_json=True, + stream=_TtyInput(), + ) + assert refused("write_note", {}) is False + assert len(prompts) == asked # JSON mode never prompts, not even on a tty + + open_engine = PolicyEngine.from_toml(DOC_ALLOW_WRITES) + nobody_asks = combined_approval( + engine=open_engine, + doc_policy=open_engine.permission_policy(tenant="default"), + flag_policy=build_policy(["*"], [], []), + tenant="default", + as_json=False, + stream=_TtyInput(), + ) + assert nobody_asks("write_note", {}) is False + assert len(prompts) == asked + + +def test_agent_combined_approval_skips_the_second_prompt_after_a_refusal(monkeypatch): + engine = PolicyEngine.from_toml(DOC_ASK_WRITES) + prompts: list[str] = [] + + def refuse(prompt: str) -> str: + prompts.append(prompt) + return "n" + + monkeypatch.setattr("builtins.input", refuse) + approve = combined_approval( + engine=engine, + doc_policy=engine.permission_policy(tenant="default"), + flag_policy=build_policy([], [], ["write_note"]), + tenant="default", + as_json=False, + stream=_TtyInput(), + ) + assert approve("write_note", {}) is False + # The document's role refused first, so the flag prompt never fired. + assert len(prompts) == 1 + assert "reviewer" in prompts[0] + + +def test_agent_document_denial_recorder_skips_flag_side_denials(): + """Recording a flag-side denial through the engine would write ALLOW + beside a refusal — so the hook records document denials only.""" + engine = PolicyEngine.from_toml(DOC_ALLOW_WRITES) + record = document_denial_recorder( + engine=engine, + doc_policy=engine.permission_policy(tenant="default"), + tenant="default", + ) + record("write_note") # the document allows it; some flag denied it + assert len(engine.audit) == 0 + + strict = PolicyEngine.from_toml(DOC_DENY_SHELL) + record_doc_deny = document_denial_recorder( + engine=strict, + doc_policy=strict.permission_policy(tenant="default"), + tenant="default", + context={"command": "agent", "run_id": "r1"}, + ) + record_doc_deny("run_command") + assert len(strict.audit) == 1 + entry = strict.audit.entries()[0] + assert entry.effect is Decision.DENY + assert entry.rule_id == "no-shell" + assert entry.tenant == "default" + assert entry.context == {"command": "agent", "run_id": "r1"} + + +def test_agent_policy_refuses_a_claude_cli_model( + tmp_path, capsys, monkeypatch, stub_tools +): + """Delegation triggers on the model, not on --executor: a claude-cli model + would hand the loop to a subprocess outside the governed harness.""" + from types import SimpleNamespace + + stub_tools() + policy = _write_doc(tmp_path / "policy.toml", DOC_ALLOW_WRITES) + monkeypatch.setattr( + "grapharc.gateway.get_model", + lambda *args, **kwargs: SimpleNamespace(_llm_type="grapharc-claude-cli"), + ) + code, payload, _ = call_json( + [ + "agent", + "t", + "--model", + "claude-cli/opus-4-6", + "--workspace", + str(tmp_path / "ws"), + "--policy", + str(policy), + ], + capsys, + ) + assert code == 2 + assert "--policy" in payload["error"] + assert "claude-cli" in payload["error"] diff --git a/tests/test_delegate_tiers.py b/tests/test_delegate_tiers.py index d80bc51..9cf7f0f 100644 --- a/tests/test_delegate_tiers.py +++ b/tests/test_delegate_tiers.py @@ -29,7 +29,7 @@ class ClaudeCliDouble(ScriptedChatModel): - """Looks like the Claude CLI backend to `_is_claude_cli`, runs nothing.""" + """Looks like the Claude CLI backend to `is_claude_cli`, runs nothing.""" @property def _llm_type(self) -> str: diff --git a/tests/test_harness_gate.py b/tests/test_harness_gate.py index 4d84294..92f5cae 100644 --- a/tests/test_harness_gate.py +++ b/tests/test_harness_gate.py @@ -22,8 +22,10 @@ import uuid import pytest +from pydantic import ValidationError from grapharc.harness import ( + CombinedPolicy, Decision, Harness, HookAction, @@ -1220,3 +1222,176 @@ def signal_self() -> str: with pytest.raises(SandboxViolation, match="signal another process"): harness.call("signal_pid", {"pid": os.getpid()}) assert harness.call("signal_self", {}) == "signalled" + + +# -- CombinedPolicy ----------------------------------------------------------- + + +def test_combined_policy_denies_when_either_side_denies(): + """Most restrictive wins: one DENY outvotes any number of allows.""" + doc = _policy([{"action": "deny", "pattern": "run_command"}]) + flags = _policy([{"action": "allow", "pattern": "*"}]) + assert CombinedPolicy(policies=[doc, flags]).decide("run_command") is Decision.DENY + # Order does not matter: there is no first match between policies. + assert CombinedPolicy(policies=[flags, doc]).decide("run_command") is Decision.DENY + + +def test_combined_policy_asks_when_either_side_asks_and_neither_denies(): + doc = _policy([{"action": "ask", "pattern": "deploy_*"}]) + flags = _policy([{"action": "allow", "pattern": "*"}]) + assert CombinedPolicy(policies=[doc, flags]).decide("deploy_prod") is Decision.ASK + # An ask on the flag side gates a document allow, the way --ask narrows. + doc_allow = _policy([{"action": "allow", "pattern": "write_*"}]) + flag_ask = _policy([{"action": "ask", "pattern": "write_note"}]) + assert ( + CombinedPolicy(policies=[doc_allow, flag_ask]).decide("write_note") + is Decision.ASK + ) + + +def test_combined_policy_allows_only_when_every_side_allows(): + doc = _policy([{"action": "allow", "pattern": "read_*"}]) + flags = PermissionPolicy(rules=[], default=Decision.ALLOW) + assert ( + CombinedPolicy(policies=[doc, flags]).decide("read_file") is Decision.ALLOW + ) + # ...and a default-deny on either side holds for unmatched tools. + strict_flags = PermissionPolicy(rules=[], default=Decision.DENY) + assert ( + CombinedPolicy(policies=[doc, strict_flags]).decide("read_file") + is Decision.DENY + ) + + +def test_combined_policy_needs_at_least_one_policy(): + with pytest.raises(ValidationError, match="at least one policy"): + CombinedPolicy(policies=[]) + + +def test_combined_policy_denies_when_its_sides_are_emptied_after_construction(): + """Defense in depth: the validator refuses an empty list, and `decide` + refuses to allow on one — so post-construction mutation fails closed.""" + combined = CombinedPolicy(policies=[_policy([{"action": "allow", "pattern": "*"}])]) + combined.policies = [] + assert combined.decide("anything") is Decision.DENY + + +def test_combined_policy_rejects_rules_written_on_the_combination_itself(): + """Own rules would be silently ignored — for a DENY that fails open.""" + side = _policy([{"action": "allow", "pattern": "*"}]) + with pytest.raises(ValidationError, match="answers from `policies`"): + CombinedPolicy( + policies=[side], + rules=[PermissionRule(action=Decision.DENY, pattern="x")], + ) + with pytest.raises(ValidationError, match="answers from `policies`"): + CombinedPolicy(policies=[side], default=Decision.ALLOW) + + +def test_combined_policy_hides_what_either_side_denies(): + """Policy-before-schema survives combination: denied means undescribed. + + The document side allows by default here, so the one hidden tool is + hidden by its explicit DENY — which a flag-side `ALLOW *` cannot widen. + """ + registry = ToolRegistry() + registry.register(ToolSpec(name="read_file", description="read", fn=_echo)) + registry.register(ToolSpec(name="run_command", description="shell", fn=_echo)) + doc = PermissionPolicy( + rules=[PermissionRule(action=Decision.DENY, pattern="run_command")], + default=Decision.ALLOW, + ) + flags = _policy([{"action": "allow", "pattern": "*"}]) + visible = registry.visible(CombinedPolicy(policies=[doc, flags])) + assert [spec.name for spec in visible] == ["read_file"] + + +def test_combined_policy_default_deny_side_is_not_widened_by_an_allow_all(): + """The opposite invariant: a default-deny document denies an unmatched + tool even beside `ALLOW *` — the default is the side's verdict, and + flags narrow a document, never widen it.""" + registry = ToolRegistry() + registry.register(ToolSpec(name="read_file", description="read", fn=_echo)) + registry.register(ToolSpec(name="run_command", description="shell", fn=_echo)) + doc = _policy([{"action": "deny", "pattern": "run_command"}]) # default DENY + flags = _policy([{"action": "allow", "pattern": "*"}]) + combined = CombinedPolicy(policies=[doc, flags]) + assert combined.decide("read_file") is Decision.DENY + assert [spec.name for spec in registry.visible(combined)] == [] + + +# -- denial hook -------------------------------------------------------------- + + +class _EchoExecutor: + """Runs everything: denials are the policy's business, not the executor's.""" + + def run(self, spec, args): + return {"ran": spec.name} + + +def _hook_harness(policy, *, approval=None): + registry = ToolRegistry() + registry.register(ToolSpec(name="write_note", description="write", fn=_echo)) + registry.register(ToolSpec(name="list_notes", description="list", fn=_echo)) + fired: list[str] = [] + harness = Harness( + registry, + policy, + executor=_EchoExecutor(), + approval=approval, + on_denial=fired.append, + ) + return harness, fired + + +def test_harness_notifies_on_denial_for_a_denied_tool(): + harness, fired = _hook_harness(_policy([{"action": "deny", "pattern": "write_*"}])) + with pytest.raises(PermissionDenied, match="denied by policy"): + harness.call("write_note", {"path": "n.txt"}) + assert fired == ["write_note"] + + +def test_harness_denial_hook_stays_silent_otherwise(): + """Allowances, granted approvals and unknown tools are not policy denials; + a refused approval already belongs to the approval path's own record.""" + harness, fired = _hook_harness( + _policy( + [ + {"action": "ask", "pattern": "write_note"}, + {"action": "allow", "pattern": "*"}, + ] + ), + approval=lambda tool, args: True, + ) + assert harness.call("list_notes", {}) == {"ran": "list_notes"} + assert harness.call("write_note", {}) == {"ran": "write_note"} + with pytest.raises(PermissionDenied, match="unknown tool"): + harness.call("no_such_tool", {}) + assert fired == [] + + refused, fired_refused = _hook_harness( + _policy( + [ + {"action": "ask", "pattern": "write_note"}, + {"action": "allow", "pattern": "*"}, + ] + ), + approval=lambda tool, args: False, + ) + with pytest.raises(PermissionDenied, match="requires approval"): + refused.call("write_note", {}) + assert fired_refused == [] + + +def test_a_failing_denial_hook_cannot_break_the_denial(): + """A broken recorder must not turn a denial into a tool error one frame up.""" + harness, _ = _hook_harness(_policy([{"action": "deny", "pattern": "*"}])) + + def broken(tool_name): + raise OSError("audit disk is gone") + + harness.on_denial = broken + with pytest.raises(PermissionDenied, match="denied by policy"): + harness.call("write_note", {}) + assert harness.denial_hook_errors == 1