# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/why-does-an-instance-of-mcp-tool-manifest-attestation-fail-at-the-registry-to-runtime-boundary-when-the-signer-identity-is-valid-but-the-tool-schema-i.md
# Remnant Warrant:    https://corpusfetch.com/c/why-does-an-instance-of-mcp-tool-manifest-attestation-fail-at-the-registry-to-runtime-boundary-when-the-signer-identity-is-valid-but-the-tool-schema-i.warrant.json
# Markdown SHA-256:   70cc04ef94527b3b0b3bea95748ebabe483d0276dba9ed076d16d6c7d0b88450
# Code AST SHA-256:   d22388b5d9908b7d498d51d732ea9d19856be25427f4fe8908bb1b108f94862d
# Verified Classes:   ManifestCanonicalizationError
# Defensive Guards:   3 fail-closed invariants (zero external deps)

import hashlib
import json
from typing import Any


class ManifestCanonicalizationError(RuntimeError):
    """Raised when the manifest dispatched to the router is not the manifest that was signed."""


def canonicalize(obj: Any) -> bytes:
    # RFC 8785 JCS-compatible for the subset we emit: sorted keys, no whitespace,
    # no NaN/Inf, integers preserved. Float formatting must be pinned by the signer.
    return json.dumps(
        obj,
        sort_keys=True,
        separators=(",", ":"),
        ensure_ascii=False,
        allow_nan=False,
    ).encode("utf-8")


def verify_dispatch_integrity(
    signed_bytes: bytes,
    signed_digest: str,
    dispatch_obj: Any,
    *,
    host_annotations_key: str = "_host_meta",
) -> None:
    """
    Enforce: canonicalize(execute_view) == canonicalize(signed_view).

    signed_bytes   : the exact bytes the registry signed and the client fetched.
    signed_digest  : sha256 hex of signed_bytes (as published in the attestation).
    dispatch_obj   : the object the router will actually validate/dispatch against.
    """
    actual_digest = hashlib.sha256(signed_bytes).hexdigest()
    if actual_digest != signed_digest:
        raise ManifestCanonicalizationError(
            f"fetched bytes do not match attestation digest: {actual_digest} != {signed_digest}"
        )

    signed_view = json.loads(signed_bytes)

    # Host annotations must live outside the signed payload; if they leaked in, fail.
    if host_annotations_key in dispatch_obj and host_annotations_key not in signed_view:
        raise ManifestCanonicalizationError(
            f"host annotation '{host_annotations_key}' present in dispatch object but not in signed manifest"
        )

    # Strip only the declared, unsigned envelope before comparison.
    compare_obj = {k: v for k, v in dispatch_obj.items() if k != host_annotations_key}

    if canonicalize(compare_obj) != canonicalize(signed_view):
        raise ManifestCanonicalizationError(
            "dispatch object diverges from signed manifest after canonicalization; "
            "attestation does not cover the executed tool"
        )