# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/why-does-an-instance-of-an-mcp-server-pass-tool-poisoning-review-but-still-leak-capability-at-runtime.md
# Remnant Warrant:    https://corpusfetch.com/c/why-does-an-instance-of-an-mcp-server-pass-tool-poisoning-review-but-still-leak-capability-at-runtime.warrant.json
# Markdown SHA-256:   4ab31b15d3ae32d0cc9308c0843c2bdb569377fe27ae263a973b0c9a3f8d6655
# Code AST SHA-256:   5b248881d012f4e0825228d4957d064bc53fb768e475fed4dfd9f7df36aaecb5
# Verified Classes:   CapabilityScopeError
# Defensive Guards:   3 fail-closed invariants (zero external deps)

class CapabilityScopeError(RuntimeError):
    """Raised when a tool's realized capability cannot be proven a subset of the granted scope."""
    pass


DISPATCH_VERBS = ("runs", "executes", "fetches", "loads", "resolves", "dispatches", "forwards")


def assert_scope_safe(tool: dict, grant_scope: str, scope_is_prefix: bool) -> None:
    """Refuse to invoke a tool whose realized capability may exceed the granted scope.

    tool: a tools/list entry with keys name, description, inputSchema, annotations.
    grant_scope: the exact resource or prefix the user consented to.
    scope_is_prefix: True if grant_scope is a directory/origin/namespace prefix.
    """
    name = tool.get("name", "<unnamed>")
    desc = (tool.get("description") or "").lower()
    schema = tool.get("inputSchema") or {}
    props = schema.get("properties") or {}

    # 1. Dispatch verbs in the description signal internal capability selection
    #    that bypasses the RPC boundary the consent layer inspects.
    hits = [v for v in DISPATCH_VERBS if v in desc]
    if hits:
        raise CapabilityScopeError(
            f"tool {name!r} description contains dispatch verbs {hits}; "
            "server may perform internal calls outside the granted scope. "
            "Require local, inspectable implementation before granting."
        )

    # 2. Open string/array/object arguments that reach a resolver are capability
    #    selectors, not data. They are the dominant leak vector.
    for arg, spec in props.items():
        t = (spec or {}).get("type")
        if t in ("string", "array", "object") and "enum" not in (spec or {}):
            if scope_is_prefix:
                raise CapabilityScopeError(
                    f"tool {name!r} argument {arg!r} is open {t!r} and grant scope "
                    f"{grant_scope!r} is a prefix. Prefix scopes require scope-"
                    "monotonicity, which requires trusting the server resolver. "
                    "Narrow grant to an exact allowlist or refuse."
                )

    # 3. Server-asserted annotations are claims, not constraints. Never infer
    #    containment from readOnlyHint alone.
    ann = tool.get("annotations") or {}
    if ann.get("readOnlyHint") and any(
        (props.get(a) or {}).get("type") in ("string", "array", "object")
        and "enum" not in (props.get(a) or {})
        for a in props
    ):
        raise CapabilityScopeError(
            f"tool {name!r} asserts readOnlyHint but has open arguments; "
            "read-only-ness is a claim about effects the client cannot verify. "
            "Treat as widening the effective grant and log it."
        )

    # 4. If we reach here, the realized surface is provably a subset of the
    #    declared surface under the stated conditions.
    return None


def guarded_invoke(tool: dict, grant_scope: str, scope_is_prefix: bool, call):
    assert_scope_safe(tool, grant_scope, scope_is_prefix)
    return call()