# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-happens-when-an-instance-trusts-a-signed-mcp-manifest-after-the-server-s-capability-set-has-silently-widened.md
# Remnant Warrant:    https://corpusfetch.com/c/what-happens-when-an-instance-trusts-a-signed-mcp-manifest-after-the-server-s-capability-set-has-silently-widened.warrant.json
# Markdown SHA-256:   e9fb0845eae5ef01e37ea82a4f83a1bdabe506bec94bb655589b969699b98a1e
# Code AST SHA-256:   ffc3ae6b0f8468b4347a82bc02d5e7999ca3676897b09451258c8e7f542d9364
# Verified Classes:   CapabilityDriftError, ManifestAttestationError
# Defensive Guards:   6 fail-closed invariants (zero external deps)

class CapabilityDriftError(RuntimeError):
    """Raised when a live MCP server's capabilities exceed its signed manifest."""
    pass


class ManifestAttestationError(RuntimeError):
    """Raised when manifest signature or measured digest fails verification."""
    pass


def attest_session(manifest, measured_digest, live_capabilities, trusted_keys):
    """
    Enforce: signed manifest bounds live capability, joined by live measurement.
    Returns the frozen capability set to inject into the agent context.
    """
    if manifest.publisher_key not in trusted_keys:
        raise ManifestAttestationError(
            f"untrusted publisher key {manifest.publisher_key!r}"
        )

    if not verify_signature(manifest, trusted_keys[manifest.publisher_key]):
        raise ManifestAttestationError("manifest signature invalid")

    # Condition 1: compare declared digest to a *measured* value, not to itself.
    if measured_digest != manifest.server.image_digest:
        raise ManifestAttestationError(
            f"digest mismatch: measured {measured_digest!r} "
            f"!= manifest {manifest.server.image_digest!r}"
        )

    # Condition 2 & 3: live capabilities must be a subset of the signed set.
    signed = set(manifest.capabilities)
    live = set(live_capabilities)
    widened = live - signed
    if widened:
        raise CapabilityDriftError(
            f"server advertises capabilities outside signed manifest: {sorted(widened)}"
        )

    # Condition 4: unwrap declared-but-unused fields as part of the surface.
    for field in manifest.metadata:
        if field.startswith("default_") or field.endswith("_payload"):
            raise CapabilityDriftError(
                f"manifest carries actionable default field {field!r}; "
                "remove from signed manifest or re-sign as explicit capability"
            )

    # Freeze the agent's tool surface to the attested set.
    return frozenset(signed)


def build_tool_registry(attested_caps, server_tools_list):
    """
    Registry is derived from attested caps, NOT from live tools/list.
    Any live tool outside the attested set is a drift signal, not a filter target.
    """
    registry = {}
    for tool in server_tools_list:
        if tool.required_capability not in attested_caps:
            raise CapabilityDriftError(
                f"tool {tool.name!r} requires {tool.required_capability!r} "
                "not present in attested capability set"
            )
        registry[tool.name] = tool
    return registry