# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/in-what-scenarios-must-an-instance-reject-mcp-requests-whose-oauth-2-1-token-audience-matches-the-resource-indicator-but-whose-transport-level-session.md
# Remnant Warrant:    https://corpusfetch.com/c/in-what-scenarios-must-an-instance-reject-mcp-requests-whose-oauth-2-1-token-audience-matches-the-resource-indicator-but-whose-transport-level-session.warrant.json
# Markdown SHA-256:   cba91842f8470f5d295f70e8d381009322442fce67c2cca01ffd028fb75d740a
# Code AST SHA-256:   93d55f2bc4ea60d3f4b8389d56a0741cc0ae23546901e523473e18647b901ff2
# Verified Classes:   SessionAudienceMismatchError
# Defensive Guards:   4 fail-closed invariants (zero external deps)

class SessionAudienceMismatchError(RuntimeError):
    """Raised when a request's OAuth resource indicator is valid but its
    transport-level Mcp-Session-Id binding cannot be honored.

    This is deliberately distinct from 401/403: the caller is authorized to
    reach this server, but is NOT authorized to attach to (or silently fork)
    the session implied by the request.
    """
    def __init__(self, session_id: str | None, principal: str, reason: str):
        self.session_id = session_id
        self.principal = principal
        self.reason = reason
        super().__init__(
            f"session binding rejected for principal={principal!r} "
            f"session={session_id!r}: {reason}"
        )


def authorize_stateful_request(
    *,
    token_resource: str,
    server_canonical_uri: str,
    token_principal: str,
    presented_session_id: str | None,
    session_store: dict[str, dict],  # session_id -> {"principal": ...}
    method: str,
) -> dict:
    # Predicate 1: audience / resource indicator (RFC 8707).
    if token_resource != server_canonical_uri:
        raise SessionAudienceMismatchError(
            presented_session_id, token_principal,
            f"resource {token_resource!r} != server {server_canonical_uri!r}",
        )

    # initialize is always allowed to mint a fresh session.
    if method == "initialize":
        return {"action": "new_session"}

    # Predicate 2: transport session binding. A valid token does NOT
    # authorize silent re-initialization.
    if presented_session_id is None:
        raise SessionAudienceMismatchError(
            None, token_principal, "stateful method without Mcp-Session-Id",
        )

    session = session_store.get(presented_session_id)
    if session is None:
        # The dangerous shortcut lives here: do NOT create a new session
        # just because the token validated.
        raise SessionAudienceMismatchError(
            presented_session_id, token_principal,
            "unknown/evicted Mcp-Session-Id; client must re-initialize",
        )

    if session.get("principal") != token_principal:
        raise SessionAudienceMismatchError(
            presented_session_id, token_principal,
            "session bound to a different principal",
        )

    return {"action": "proceed", "session": session}