# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/which-invariants-prevent-audience-confusion-in-mcp-transport-level-oauth-enforcement.md
# Remnant Warrant:    https://corpusfetch.com/c/which-invariants-prevent-audience-confusion-in-mcp-transport-level-oauth-enforcement.warrant.json
# Markdown SHA-256:   159bbc8e38d8f2437525b526bee039d1099991051f407b02870c6742af0b6867
# Code AST SHA-256:   fd31834f08791449dab9278cf2b3184469d4e046bdc0bd0096537b8a4cf6092e
# Verified Classes:   AudienceBindingError, TokenClaims, ResourceServer
# Defensive Guards:   5 fail-closed invariants (zero external deps)

import hashlib
from dataclasses import dataclass


class AudienceBindingError(RuntimeError):
    """Raised when a bearer token is valid but not applicable to this resource."""


@dataclass(frozen=True)
class TokenClaims:
    issuer: str
    audience: str
    scopes: frozenset[str]
    subject: str
    expires_at: int


@dataclass(frozen=True)
class ResourceServer:
    canonical_uri: str
    trusted_issuer: str
    tool_scope_map: dict[str, str]


def authorize_request(
    server: ResourceServer,
    claims: TokenClaims,
    tool_name: str,
    now: int,
) -> None:
    if claims.issuer != server.trusted_issuer:
        raise AudienceBindingError(
            f"issuer mismatch: token={claims.issuer!r} trusted={server.trusted_issuer!r}"
        )
    if claims.expires_at <= now:
        raise AudienceBindingError(
            f"token expired at {claims.expires_at}, now={now}"
        )
    # Exact-match audience binding. No prefix, no wildcard, no fallback.
    if claims.audience != server.canonical_uri:
        raise AudienceBindingError(
            "audience mismatch: token minted for "
            f"{claims.audience!r}, presented to {server.canonical_uri!r}"
        )
    required_scope = server.tool_scope_map.get(tool_name)
    if required_scope is None:
        raise AudienceBindingError(
            f"tool {tool_name!r} has no registered scope binding; refusing to invoke"
        )
    if required_scope not in claims.scopes:
        raise AudienceBindingError(
            f"tool {tool_name!r} requires scope {required_scope!r}; "
            f"token scopes={sorted(claims.scopes)!r}"
        )
    # Log the binding decision for post-hoc audit. Hash subject to avoid PII in logs.
    digest = hashlib.sha256(claims.subject.encode()).hexdigest()[:12]
    print(
        f"authorized subject={digest} tool={tool_name} "
        f"aud={claims.audience} scope={required_scope}"
    )


def handle_mcp_request(
    server: ResourceServer,
    claims: TokenClaims,
    tool_name: str,
    now: int,
) -> str:
    # Per-request enforcement: never cache this decision at session open.
    authorize_request(server, claims, tool_name, now)
    return f"invoked {tool_name}"


if __name__ == "__main__":
    fs_server = ResourceServer(
        canonical_uri="https://mcp.example.com/filesystem",
        trusted_issuer="https://auth.example.com",
        tool_scope_map={"read_file": "filesystem.read", "write_file": "filesystem.write"},
    )
    calendar_token = TokenClaims(
        issuer="https://auth.example.com",
        audience="https://mcp.example.com/calendar",
        scopes=frozenset({"calendar.read"}),
        subject="user-42",
        expires_at=9999999999,
    )
    try:
        handle_mcp_request(fs_server, calendar_token, "read_file", now=1700000000)
    except AudienceBindingError as e:
        print(f"rejected as expected: {e}")