"""The operator's own tool allowlist, enforced independently of any server. There are two parties who get a say in which tools a run may use, and they are not the same party: * the **server** proposes, per run, through ``JobSpec.allowed_tools``; * the **operator** approves, once, through this module. A tool runs only if both agree and it is actually installed. The operator's list is the one that cannot be widened remotely: a compromised or simply mistaken server can name any tool it likes and will be refused, because the worker checks its own list before it checks the job's. This is the same argument as PLAN.md ยง4.7 about per-server credential scoping, applied to capability rather than credentials. **The default is to approve nothing.** A freshly installed agent advertises no tools and every call fails closed. That is not a degraded state; it is what keeps ``distinct_tools``' claim in ``__init__`` literally true until an operator decides otherwise. Approving the local bundle is one flag. Approving anything that reaches the network is a separate, second decision, because the runtime's no-outbound-request property is exactly what it costs. """ from __future__ import annotations import json import os from collections.abc import Iterable, Mapping, Sequence from dataclasses import dataclass, field from pathlib import Path from .core import ToolRef, ToolSpec #: Comma-separated ``id@version`` refs, or the bundle name ``local``. APPROVED_TOOLS_ENV_VAR = "DISTINCT_APPROVED_TOOLS" #: Comma-separated ``id@version`` refs additionally cleared to reach the #: network. Being in this list does nothing on its own; a tool must also be #: in the approved list. APPROVED_EGRESS_ENV_VAR = "DISTINCT_APPROVED_EGRESS_TOOLS" #: Path to a JSON policy file, which takes precedence over the variables. POLICY_FILE_ENV_VAR = "DISTINCT_TOOL_POLICY" #: Name for "every local tool that declares no hosts". Expanded against the #: installed registry at load time, so it can never silently come to include a #: networked tool added later: expansion filters on ``required_hosts``. LOCAL_BUNDLE = "local" MAX_APPROVED_TOOLS = 64 class ToolPolicyError(ValueError): """A policy could not be read. Callers must fail closed, not continue.""" def _parse_ref(value: object) -> ToolRef: if not isinstance(value, str) or value.count("@") != 1: raise ToolPolicyError( f"approved tool {value!r} must use the exact id@version form, for example calculate@1" ) tool_id, version = value.split("@", 1) try: return ToolRef(tool_id.strip(), version.strip()) except (TypeError, ValueError) as exc: raise ToolPolicyError(f"approved tool {value!r} is malformed: {exc}") from exc def _split(value: object) -> tuple[str, ...]: if value is None: return () if isinstance(value, str): return tuple(item.strip() for item in value.split(",") if item.strip()) if isinstance(value, Sequence): return tuple(str(item).strip() for item in value if str(item).strip()) raise ToolPolicyError("tool policy entries must be a list or a comma-separated string") @dataclass(frozen=True) class OperatorPolicy: """Which exact tool versions this worker's operator has approved. ``approved`` is exhaustive: a ref absent from it is refused, whatever the registry contains and whatever a server asks for. ``egress_approved`` is a second gate that only matters for tools declaring ``required_hosts``. """ approved: frozenset[ToolRef] = frozenset() egress_approved: frozenset[ToolRef] = frozenset() source: str = "default (nothing approved)" rejected: tuple[str, ...] = field(default_factory=tuple) def __post_init__(self) -> None: approved = frozenset(self.approved) if len(approved) > MAX_APPROVED_TOOLS: raise ToolPolicyError( f"at most {MAX_APPROVED_TOOLS} tools may be approved; " "a large local catalogue costs selection accuracy on small models" ) object.__setattr__(self, "approved", approved) egress = frozenset(self.egress_approved) stray = egress.difference(approved) if stray: # Clearing a tool for the network without approving the tool is # almost certainly a mistake, and guessing which half was meant # would be guessing in the permissive direction. names = ", ".join(sorted(f"{ref.tool_id}@{ref.version}" for ref in stray)) raise ToolPolicyError( f"egress approval names tools that are not approved at all: {names}" ) object.__setattr__(self, "egress_approved", egress) object.__setattr__(self, "rejected", tuple(str(item)[:200] for item in self.rejected)) def permits(self, ref: ToolRef) -> bool: return ref in self.approved def permits_egress(self, ref: ToolRef) -> bool: return ref in self.egress_approved def refusal_reason(self, spec: ToolSpec) -> str | None: """Return why ``spec`` may not run, or ``None`` if it may. Separate from :meth:`permits` because the reason is what the operator and the user need; a bare boolean turns two different problems into one indistinguishable failure. """ ref = spec.ref if ref not in self.approved: return ( f"tool {ref.tool_id}@{ref.version} is not on this worker's approved list; " "the operator must approve it before any run can use it" ) if spec.required_hosts and ref not in self.egress_approved: hosts = ", ".join(sorted(spec.required_hosts)) return ( f"tool {ref.tool_id}@{ref.version} needs outbound access to {hosts} and this " "worker has not approved network access for it" ) return None def permitted_specs(self, specs: Iterable[ToolSpec]) -> tuple[ToolSpec, ...]: """Filter installed specs down to what may actually be advertised.""" return tuple( spec for spec in sorted(specs, key=lambda item: (item.tool_id, item.version)) if self.refusal_reason(spec) is None ) def advertised_refs(self, specs: Iterable[ToolSpec]) -> tuple[str, ...]: return tuple(f"{spec.tool_id}@{spec.version}" for spec in self.permitted_specs(specs)) def describe(self, specs: Iterable[ToolSpec] = ()) -> str: """One block of text for the agent console at start-up. Says what is on, what is off and what was thrown out, because an operator who cannot see the effective policy cannot check it. """ # Materialised once: ``specs`` may be a generator, and walking it twice # would silently report an empty second half. installed = tuple(specs) permitted = self.permitted_specs(installed) lines = [f"Tool approval policy: {self.source}"] if not permitted: lines.append(" No tools are approved. Every tool call will fail closed.") for spec in permitted: marker = "network" if spec.required_hosts else "local, no egress" lines.append(f" {spec.tool_id}@{spec.version} ({marker})") for spec in sorted(installed, key=lambda item: (item.tool_id, item.version)): reason = self.refusal_reason(spec) if reason is not None: lines.append(f" withheld: {spec.tool_id}@{spec.version}: {reason}") for item in self.rejected: lines.append(f" ignored malformed policy entry: {item}") return "\n".join(lines) #: Approves nothing. The value a caller should fall back to on any error. DENY_ALL = OperatorPolicy() def _expand( names: Sequence[str], installed: Sequence[ToolSpec], *, rejected: list[str], ) -> set[ToolRef]: local_refs = {spec.ref for spec in installed if not spec.required_hosts} refs: set[ToolRef] = set() for name in names: if name == LOCAL_BUNDLE: # Expanded against what is installed *and* declares no hosts, so # the bundle cannot quietly grow a networked member. refs.update(local_refs) continue try: refs.add(_parse_ref(name)) except ToolPolicyError as exc: rejected.append(str(exc)) return refs def load_policy( installed: Iterable[ToolSpec] = (), *, environ: Mapping[str, str] | None = None, ) -> OperatorPolicy: """Read the operator's policy, or return :data:`DENY_ALL`. Precedence: the policy file named by :data:`POLICY_FILE_ENV_VAR`, else the two environment variables, else nothing approved. Every failure path ends at ``DENY_ALL`` with a reason in ``source``: a policy that could not be read must never be treated as a permissive one. """ values = os.environ if environ is None else environ specs = tuple(installed) rejected: list[str] = [] path_value = values.get(POLICY_FILE_ENV_VAR, "").strip() if path_value: try: raw = Path(path_value).expanduser().read_text(encoding="utf-8") document = json.loads(raw) except (OSError, ValueError) as exc: return OperatorPolicy( source=f"{path_value} could not be read ({type(exc).__name__}); nothing approved" ) if not isinstance(document, Mapping): return OperatorPolicy(source=f"{path_value} is not a JSON object; nothing approved") try: approved = _expand(_split(document.get("approved")), specs, rejected=rejected) egress = _expand(_split(document.get("egress")), specs, rejected=rejected) return OperatorPolicy( approved=frozenset(approved), egress_approved=frozenset(egress.intersection(approved)), source=f"{path_value}", rejected=tuple(rejected), ) except ToolPolicyError as exc: return OperatorPolicy(source=f"{path_value} was rejected ({exc}); nothing approved") approved_names = _split(values.get(APPROVED_TOOLS_ENV_VAR)) if not approved_names: return OperatorPolicy() egress_names = _split(values.get(APPROVED_EGRESS_ENV_VAR)) try: approved = _expand(approved_names, specs, rejected=rejected) egress = _expand(egress_names, specs, rejected=rejected) return OperatorPolicy( approved=frozenset(approved), egress_approved=frozenset(egress.intersection(approved)), source=f"{APPROVED_TOOLS_ENV_VAR}", rejected=tuple(rejected), ) except ToolPolicyError as exc: return OperatorPolicy(source=f"{APPROVED_TOOLS_ENV_VAR} was rejected ({exc}); nothing approved") __all__ = [ "APPROVED_EGRESS_ENV_VAR", "APPROVED_TOOLS_ENV_VAR", "DENY_ALL", "LOCAL_BUNDLE", "MAX_APPROVED_TOOLS", "POLICY_FILE_ENV_VAR", "OperatorPolicy", "ToolPolicyError", "load_policy", ]