distinct_tools/approval.py

"""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",
]