# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/which-invariants-prevent-agent-identity-binding-from-silently-degrading-under-mcp-session-reuse.md
# Remnant Warrant:    https://corpusfetch.com/c/which-invariants-prevent-agent-identity-binding-from-silently-degrading-under-mcp-session-reuse.warrant.json
# Markdown SHA-256:   b71896ce5544bed40ee2b2011474fad538ace6f61572375249e766f3cd4a31ef
# Code AST SHA-256:   f8220b7ed597f6d41d0388ee76ca358fa1a5f78e40c376a47d882a74fd40abd0
# Verified Classes:   SessionIdentityMismatch, ProofOfPossessionError, Principal, Session
# Defensive Guards:   5 fail-closed invariants (zero external deps)

import hashlib
import hmac
import time
from dataclasses import dataclass, field
from typing import Optional


class SessionIdentityMismatch(RuntimeError):
    """Raised when a request's proven principal does not match the session's bound principal."""


class ProofOfPossessionError(RuntimeError):
    """Raised when the per-request proof-of-possession artifact is missing or invalid."""


@dataclass
class Principal:
    agent_id: str
    public_key: bytes
    scopes: frozenset


@dataclass
class Session:
    session_id: str
    bound_principal_id: str
    created_at: float
    # NOTE: no scopes cached here on purpose; authorization is re-derived per request.


def _verify_pop(principal: Principal, request_body: bytes, signature: bytes) -> bool:
    # DPoP-style: signature over the request body with the principal's registered key.
    expected = hmac.new(principal.public_key, request_body, hashlib.sha256).digest()
    return hmac.compare_digest(expected, signature)


def authorize_tool_call(
    sessions: dict,
    principals: dict,
    session_id: Optional[str],
    request_body: bytes,
    signature: Optional[bytes],
    required_scope: str,
    now: Optional[float] = None,
) -> Principal:
    now = now if now is not None else time.time()

    if session_id is None or signature is None:
        raise ProofOfPossessionError("missing session id or proof-of-possession artifact")

    session = sessions.get(session_id)
    if session is None:
        raise SessionIdentityMismatch(f"unknown session {session_id!r}")

    # Resolve principal from the PoP key, not from the session object.
    proven_principal: Optional[Principal] = None
    for p in principals.values():
        if _verify_pop(p, request_body, signature):
            proven_principal = p
            break

    if proven_principal is None:
        raise ProofOfPossessionError("proof-of-possession did not verify against any registered key")

    # Continuous binding check: the proven principal must be the one the session was bound to.
    if proven_principal.agent_id != session.bound_principal_id:
        raise SessionIdentityMismatch(
            f"session {session_id!r} bound to {session.bound_principal_id!r} "
            f"but request proven by {proven_principal.agent_id!r}"
        )

    # Re-evaluate authorization against current policy, not a snapshot.
    if required_scope not in proven_principal.scopes:
        raise SessionIdentityMismatch(
            f"principal {proven_principal.agent_id!r} lacks scope {required_scope!r} at {now:.0f}"
        )

    return proven_principal