# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-structural-signals-indicate-a-capability-token-has-been-silently-downgraded-across-an-ebpf-syscall-interception-boundary.md
# Remnant Warrant:    https://corpusfetch.com/c/what-structural-signals-indicate-a-capability-token-has-been-silently-downgraded-across-an-ebpf-syscall-interception-boundary.warrant.json
# Markdown SHA-256:   3bed12cca0ed9f613a3dc599427164c0e2f057ffd133e762d849fd6f5d820f93
# Code AST SHA-256:   3934de8608d9e01eda1ef08228fd49d8538c45b1cf53e08345e9e31d65075f19
# Verified Classes:   CapabilityDowngradeError
# Defensive Guards:   3 fail-closed invariants (zero external deps)

import hashlib
import hmac


class CapabilityDowngradeError(RuntimeError):
    """Raised when the capability payload enforced at the syscall hook
    does not match the capability set bound into the token's signed envelope."""


def _capability_hash(capabilities: dict) -> bytes:
    canonical = repr(sorted(capabilities.items())).encode("utf-8")
    return hashlib.sha256(canonical).digest()


def enforce_capability_token(
    envelope: dict,
    enforced_capabilities: dict,
    signing_key: bytes,
) -> None:
    """
    Validate a capability token at an eBPF-style syscall interception hook.

    The envelope carries authenticated metadata (subject, expiry, signature)
    plus a capability_hash that binds the *intended* capability set. The
    enforced_capabilities are the set actually being applied to the syscall,
    which may have been sourced from a stale cache.

    Raises CapabilityDowngradeError if the two disagree.
    """
    # 1. Verify the envelope signature over its authenticated fields.
    signed_fields = f"{envelope['subject']}:{envelope['expires_at']}:{envelope['capability_hash']}"
    expected_sig = hmac.new(
        signing_key, signed_fields.encode("utf-8"), hashlib.sha256
    ).hexdigest()
    if not hmac.compare_digest(expected_sig, envelope["signature"]):
        raise CapabilityDowngradeError(
            "Envelope signature invalid; token cannot be trusted."
        )

    # 2. Recompute the capability hash from what is *actually being enforced*.
    observed_hash = _capability_hash(enforced_capabilities)

    # 3. The invariant: enforced capabilities must match the bound hash.
    if not hmac.compare_digest(observed_hash.hex(), envelope["capability_hash"]):
        raise CapabilityDowngradeError(
            "Capability set enforced at hook does not match token envelope. "
            f"bound={envelope['capability_hash'][:16]} "
            f"observed={observed_hash.hex()[:16]}. "
            "Token was silently downgraded across the interception boundary; "
            "abort syscall and re-issue token from a clean cache."
        )

    # 4. Expiry check (after integrity, since expiry is inside the signature).
    if envelope["expires_at"] < envelope["now"]:
        raise CapabilityDowngradeError("Token expired; refusing syscall.")


def intercept_write_syscall(envelope: dict, cache: dict, signing_key: bytes) -> None:
    # Simulate a cache that returns a stale, narrower capability set.
    enforced = cache.get("capabilities", {"write_quota": 4096})
    enforce_capability_token(envelope, enforced, signing_key)