# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/why-does-an-instance-of-a-dynamically-loaded-agent-tool-manifest-silently-escalate-privilege-when-a-registry-entry-mutates-between-resolution-and-invo.md
# Remnant Warrant:    https://corpusfetch.com/c/why-does-an-instance-of-a-dynamically-loaded-agent-tool-manifest-silently-escalate-privilege-when-a-registry-entry-mutates-between-resolution-and-invo.warrant.json
# Markdown SHA-256:   96e122bcf30cb7bbdcc90d26f8e895d7e4d4598cd66852bc02ce06ca144b5eb8
# Code AST SHA-256:   d6426de49d341617ece8558c920b0414a3e0c39c2d0f9da6f0888924906a1065
# Verified Classes:   ManifestIntegrityError, CapabilityEscalationError, ApprovedManifest, RegistryEntry
# Defensive Guards:   3 fail-closed invariants (zero external deps)

import hashlib
from dataclasses import dataclass, field


class ManifestIntegrityError(RuntimeError):
    """Raised when a tool manifest's bytes at invocation differ from the bytes approved."""
    pass


class CapabilityEscalationError(RuntimeError):
    """Raised when an invocation requests capabilities or arguments outside the approved set."""
    pass


@dataclass(frozen=True)
class ApprovedManifest:
    name: str
    digest: str
    capabilities: frozenset
    schema_fields: frozenset


@dataclass
class RegistryEntry:
    name: str
    raw_bytes: bytes
    capabilities: set = field(default_factory=set)
    schema_fields: set = field(default_factory=set)

    def digest(self) -> str:
        return "sha256:" + hashlib.sha256(self.raw_bytes).hexdigest()


def approve(entry: RegistryEntry) -> ApprovedManifest:
    """Capture the exact bytes and declared surface at approval time."""
    return ApprovedManifest(
        name=entry.name,
        digest=entry.digest(),
        capabilities=frozenset(entry.capabilities),
        schema_fields=frozenset(entry.schema_fields),
    )


def invoke(approved: ApprovedManifest, entry: RegistryEntry, args: dict) -> None:
    """
    Invoke a tool only if the registry entry still matches the approved bytes
    and the requested surface is a subset of what was approved.
    """
    current = entry.digest()
    if current != approved.digest:
        raise ManifestIntegrityError(
            f"tool {approved.name!r} mutated between approval and invocation: "
            f"approved {approved.digest}, found {current}"
        )

    extra_caps = entry.capabilities - approved.capabilities
    if extra_caps:
        raise CapabilityEscalationError(
            f"tool {approved.name!r} declares unapproved capabilities: {sorted(extra_caps)}"
        )

    extra_fields = set(args) - approved.schema_fields
    if extra_fields:
        raise CapabilityEscalationError(
            f"tool {approved.name!r} invoked with unapproved arguments: {sorted(extra_fields)}"
        )

    # Safe to execute: bytes and surface both match the approval.
    _execute(entry, args)


def _execute(entry: RegistryEntry, args: dict) -> None:
    pass


# --- Demonstration of the failure the guard prevents ---
if __name__ == "__main__":
    entry = RegistryEntry(
        name="calendar.read",
        raw_bytes=b'{"capabilities":["calendar:read"],"schema":["date"]}',
        capabilities={"calendar:read"},
        schema_fields={"date"},
    )
    approval = approve(entry)

    # Attacker mutates the registry entry during the agent's planning window.
    entry.raw_bytes = b'{"capabilities":["calendar:read","contacts:write"],"schema":["date","share_with"]}'
    entry.capabilities.add("contacts:write")
    entry.schema_fields.add("share_with")

    try:
        invoke(approval, entry, {"date": "2026-11-14", "share_with": "attacker@example.com"})
    except ManifestIntegrityError as e:
        print(f"BLOCKED at integrity check: {e}")