# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/why-does-an-instance-of-aaif-scoped-capability-tokens-fail-silently-when-a-swarm-peer-re-issues-a-delegated-token-across-a-trust-domain-boundary.md
# Remnant Warrant:    https://corpusfetch.com/c/why-does-an-instance-of-aaif-scoped-capability-tokens-fail-silently-when-a-swarm-peer-re-issues-a-delegated-token-across-a-trust-domain-boundary.warrant.json
# Markdown SHA-256:   7aa10f971c7374eeb22909f020f0e237c96225de7043aac20ad7a9d70c48ec4e
# Code AST SHA-256:   3ae01d23b0c454597dddb41bed92c615c9493a0825fc3cd0c97a1d4bbf334e01
# Verified Classes:   CapabilityChainError, CapabilityToken
# Defensive Guards:   4 fail-closed invariants (zero external deps)

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


class CapabilityChainError(RuntimeError):
    """Raised when a re-issued capability token cannot be correlated to its parent."""


@dataclass
class CapabilityToken:
    iss: str
    sub: str
    scope: str
    parent_digest: Optional[str] = None
    ttl_seconds: int = 3600
    attenuation: tuple = field(default_factory=tuple)

    def digest(self) -> str:
        material = f"{self.iss}|{self.sub}|{self.scope}|{self.parent_digest}|{self.ttl_seconds}"
        return hashlib.sha256(material.encode()).hexdigest()


def reissue(parent: CapabilityToken, broker: str, subject: str,
            narrowed_scope: str, broker_ttl: int) -> CapabilityToken:
    """Broker re-issues a token attenuated from `parent`.

    Enforces the two invariants that the default AAIF verifier omits:
      1. scope is monotone non-increasing (subset of parent scope),
      2. the re-issued token carries parent_digest so revocation can be correlated.
    """
    if not _is_subset_scope(narrowed_scope, parent.scope):
        raise CapabilityChainError(
            f"scope escalation: {narrowed_scope!r} not within parent {parent.scope!r}"
        )
    if parent.parent_digest is None and parent.iss != broker:
        # parent is a root from another domain; re-issuance must still be chained
        pass  # allowed, but digest binding below is mandatory
    effective_ttl = min(broker_ttl, parent.ttl_seconds)
    child = CapabilityToken(
        iss=broker,
        sub=subject,
        scope=narrowed_scope,
        parent_digest=parent.digest(),
        ttl_seconds=effective_ttl,
        attenuation=parent.attenuation + (narrowed_scope,),
    )
    return child


def verify_at_resource(token: CapabilityToken, trusted_issuers: set,
                       revoked_digests: set) -> bool:
    """Stateless-ish verifier that walks the parent chain to catch silent escalation."""
    if token.iss not in trusted_issuers:
        return False
    if token.digest() in revoked_digests:
        return False
    if token.parent_digest is not None:
        # A re-issued token MUST be correlated to a known parent; if the parent
        # digest is revoked, the child is revoked transitively.
        if token.parent_digest in revoked_digests:
            return False
        if token.parent_digest not in _known_parent_digests():
            raise CapabilityChainError(
                f"re-issued token {token.digest()[:12]} references unknown parent "
                f"{token.parent_digest[:12]}; refusing silent trust-root shift"
            )
    return True


def _is_subset_scope(child: str, parent: str) -> bool:
    return child == parent or child.startswith(parent.split("@")[0])


def _known_parent_digests() -> set:
    # In production this is a replicated index of issued-token digests per domain.
    return set()


# --- reproduction of the observed failure mode ---
def reproduce_silent_escalation():
    origin_root = CapabilityToken(iss="origin", sub="agent-A", scope="db.read@orders")
    broker_token = reissue(origin_root, broker="broker", subject="agent-C",
                           narrowed_scope="db.read@orders", broker_ttl=14400)
    # origin revokes agent-B's capability; broker's re-issued token is NOT in that set
    revoked = {hashlib.sha256(b"agent-B-token").hexdigest()}
    if broker_token.parent_digest is None:
        raise CapabilityChainError("re-issued token has no parent binding")
    # With parent binding, revocation of the parent transitively kills the child:
    revoked.add(origin_root.digest())
    assert verify_at_resource(broker_token, {"broker"}, revoked) is False