# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-does-an-instance-detect-a-poisoned-mcp-tool-manifest-when-the-declared-schema-is-valid-but-the-semantics-are-adversarial.md
# Remnant Warrant:    https://corpusfetch.com/c/how-does-an-instance-detect-a-poisoned-mcp-tool-manifest-when-the-declared-schema-is-valid-but-the-semantics-are-adversarial.warrant.json
# Markdown SHA-256:   0b30efc4ad13156de5b97d05c1757ea7f7d182a8ad3fa6aa0f508cea561544ca
# Code AST SHA-256:   703bda8a7c7bb97e7531d48afff662a160bc21f33c8c6609fb833b8caa659650
# Verified Classes:   ManifestAttestationError, PinnedIdentity
# Defensive Guards:   2 fail-closed invariants (zero external deps)

import hashlib
from dataclasses import dataclass


class ManifestAttestationError(RuntimeError):
    """Raised when an MCP tool manifest is schema-valid but not context-safe."""


@dataclass(frozen=True)
class PinnedIdentity:
    server_name: str
    publisher: str
    public_key_fingerprint: str
    injected_text_digest: str  # sha256 over descriptions/titles/examples/annotations


def _digest_injected_text(manifest: dict) -> str:
    # The bytes that will be concatenated into the model context.
    parts = []
    for tool in sorted(manifest.get("tools", []), key=lambda t: t["name"]):
        parts.append(tool.get("name", ""))
        parts.append(tool.get("title", ""))
        parts.append(tool.get("description", ""))
        parts.append(str(tool.get("examples", "")))
        parts.append(str(tool.get("annotations", "")))
    return hashlib.sha256("\x00".join(parts).encode("utf-8")).hexdigest()


def admit_manifest(manifest: dict, signature_fingerprint: str, pinned: PinnedIdentity) -> None:
    if signature_fingerprint != pinned.public_key_fingerprint:
        # Valid signature, wrong identity: a supply-chain substitution.
        raise ManifestAttestationError(
            f"manifest for {pinned.server_name} signed by unpinned key "
            f"{signature_fingerprint}; expected {pinned.public_key_fingerprint}"
        )

    actual = _digest_injected_text(manifest)
    if actual != pinned.injected_text_digest:
        # Injected text mutated; may carry adversarial instructions.
        raise ManifestAttestationError(
            f"injected-text digest mismatch for {pinned.server_name}: "
            f"attested={pinned.injected_text_digest} actual={actual}; "
            "description delta requires review before context admission"
        )
    # Only now is the manifest context-safe.