# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/by-what-cryptographic-and-provenance-gates-must-an-instance-authenticate-inter-agent-handoffs-to-prevent-confused-deputy-escalation.md
# Remnant Warrant:    https://corpusfetch.com/c/by-what-cryptographic-and-provenance-gates-must-an-instance-authenticate-inter-agent-handoffs-to-prevent-confused-deputy-escalation.warrant.json
# Markdown SHA-256:   9e633edf5775a8811f586e98e8cf462b8dddeed6210e6abe70bf892780e68ed2
# Code AST SHA-256:   d4f68e4edabc4c6dbf2c353472c76bebb002ae784030d989ada6e35ba72a7fee
# Verified Classes:   ConfusedDeputyEscalationError, SignedAgentHandoffEnvelope, VerifiedHandoffContext, InterAgentHandoffGuard
# Defensive Guards:   5 fail-closed invariants (zero external deps)

import hashlib
import hmac
from dataclasses import dataclass


class ConfusedDeputyEscalationError(PermissionError):
    '''Raised when an inter-agent handoff fails MAC authentication, replays a nonce, or escalates scope.'''
    pass


@dataclass(frozen=True)
class SignedAgentHandoffEnvelope:
    handoff_id: str
    origin_principal_id: str
    parent_agent_id: str
    nonce: str
    expires_at_epoch: float
    requested_scopes: frozenset[str]
    hmac_hex: str


@dataclass(frozen=True)
class VerifiedHandoffContext:
    handoff_id: str
    origin_principal_id: str
    effective_attenuated_scopes: frozenset[str]


class InterAgentHandoffGuard:
    '''
    Prevents confused-deputy privilege escalation across multi-agent handoffs:
    1. Verifies HMAC-SHA256 authenticity over canonical envelope fields via `hmac.compare_digest`.
    2. Blocks expired tokens and replayed nonces.
    3. Computes `effective_scopes = origin_acl & parent_scopes & requested_scopes`
       (never granting the receiving deputy's ambient privileges).
    '''

    def __init__(self, broker_secret_key: bytes) -> None:
        if len(broker_secret_key) < 16:
            raise ValueError("Broker secret key must be at least 16 bytes.")
        self._key = broker_secret_key
        self._consumed_nonces: set[str] = set()

    @staticmethod
    def _canonical_message(
        handoff_id: str,
        origin_principal_id: str,
        parent_agent_id: str,
        nonce: str,
        expires_at_epoch: float,
        requested_scopes: frozenset[str],
    ) -> bytes:
        scope_str = ",".join(sorted(requested_scopes))
        raw = f"{handoff_id}|{origin_principal_id}|{parent_agent_id}|{nonce}|{expires_at_epoch:.3f}|{scope_str}"
        return raw.encode("utf-8")

    def mint_envelope(
        self,
        *,
        handoff_id: str,
        origin_principal_id: str,
        parent_agent_id: str,
        nonce: str,
        expires_at_epoch: float,
        requested_scopes: frozenset[str],
    ) -> SignedAgentHandoffEnvelope:
        msg = self._canonical_message(
            handoff_id, origin_principal_id, parent_agent_id, nonce, expires_at_epoch, requested_scopes
        )
        sig = hmac.new(self._key, msg, hashlib.sha256).hexdigest()
        return SignedAgentHandoffEnvelope(
            handoff_id=handoff_id,
            origin_principal_id=origin_principal_id,
            parent_agent_id=parent_agent_id,
            nonce=nonce,
            expires_at_epoch=expires_at_epoch,
            requested_scopes=requested_scopes,
            hmac_hex=sig,
        )

    def authenticate_and_attenuate(
        self,
        envelope: SignedAgentHandoffEnvelope,
        *,
        origin_principal_acl: frozenset[str],
        parent_agent_scopes: frozenset[str],
        required_tool_scopes: frozenset[str],
        now_epoch: float,
    ) -> VerifiedHandoffContext:
        # Gate 1: Constant-time HMAC verification
        expected_msg = self._canonical_message(
            envelope.handoff_id,
            envelope.origin_principal_id,
            envelope.parent_agent_id,
            envelope.nonce,
            envelope.expires_at_epoch,
            envelope.requested_scopes,
        )
        expected_sig = hmac.new(self._key, expected_msg, hashlib.sha256).hexdigest()
        if not hmac.compare_digest(expected_sig, envelope.hmac_hex):
            raise ConfusedDeputyEscalationError(
                f"Handoff '{envelope.handoff_id}' rejected: invalid cryptographic HMAC signature."
            )

        # Gate 2: Expiration and single-use nonce replay check
        if now_epoch > envelope.expires_at_epoch:
            raise ConfusedDeputyEscalationError(
                f"Handoff '{envelope.handoff_id}' rejected: token expired at {envelope.expires_at_epoch}."
            )
        if envelope.nonce in self._consumed_nonces:
            raise ConfusedDeputyEscalationError(
                f"Handoff '{envelope.handoff_id}' rejected: replay detected for nonce '{envelope.nonce}'."
            )
        self._consumed_nonces.add(envelope.nonce)

        # Gate 3: Strict capability attenuation intersection (zero ambient escalation)
        effective_scopes = origin_principal_acl & parent_agent_scopes & envelope.requested_scopes

        # Gate 4: Verify tool scopes are within the attenuated intersection
        if not required_tool_scopes.issubset(effective_scopes):
            missing = sorted(required_tool_scopes - effective_scopes)
            raise ConfusedDeputyEscalationError(
                f"Confused Deputy Escalation Blocked on '{envelope.handoff_id}': tool requires {missing}, "
                f"but attenuated intersection for principal '{envelope.origin_principal_id}' only grants {sorted(effective_scopes)}."
            )

        return VerifiedHandoffContext(
            handoff_id=envelope.handoff_id,
            origin_principal_id=envelope.origin_principal_id,
            effective_attenuated_scopes=effective_scopes,
        )