# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-does-an-instance-detect-a-capability-escalating-skill-injection-when-manifest-signatures-verify-but-the-signed-manifest-is-a-valid-superset-of-the.md
# Remnant Warrant:    https://corpusfetch.com/c/how-does-an-instance-detect-a-capability-escalating-skill-injection-when-manifest-signatures-verify-but-the-signed-manifest-is-a-valid-superset-of-the.warrant.json
# Markdown SHA-256:   8ff680e34a132f4a2225669ee9570653fa14c99f0f7384cdf1c50669348067d0
# Code AST SHA-256:   d53876a6241dd374705aeefba17559a3fdcc1ce043dceafdff8b417255da3d82
# Verified Classes:   CapabilityEscalationError, Capability
# Defensive Guards:   4 fail-closed invariants (zero external deps)

from dataclasses import dataclass
from typing import FrozenSet


class CapabilityEscalationError(RuntimeError):
    """Raised when a signed manifest is a capability escalation over the baseline."""


@dataclass(frozen=True)
class Capability:
    authority_class: str      # e.g. "fs", "net", "exec"
    scope: str                # e.g. "/data/reports", "*", "api.example.com"
    transitive: bool = False  # True if exercising it can reach other classes

    def reaches(self) -> FrozenSet[str]:
        # Transitive capabilities subsume every class; non-transitive touch only their own.
        if self.transitive:
            return frozenset({"fs", "net", "exec", "env", "proc"})
        return frozenset({self.authority_class})


def _expand_transitive(caps: FrozenSet[Capability]) -> FrozenSet[str]:
    classes = set()
    for c in caps:
        classes |= c.reaches()
    return frozenset(classes)


def detect_escalation(
    baseline: FrozenSet[Capability],
    incoming: FrozenSet[Capability],
) -> None:
    """Raise CapabilityEscalationError if `incoming` escalates over `baseline`.

    Escalation is defined as: an incoming capability whose authority class is
    not already reachable from the baseline, OR whose scope is not a subset of
    the baseline scopes in that class, OR a transitive capability whose closure
    exceeds the baseline's reachable classes.
    """
    baseline_classes = _expand_transitive(baseline)
    baseline_scopes = {}
    for c in baseline:
        baseline_scopes.setdefault(c.authority_class, set()).add(c.scope)

    for cap in incoming:
        cap_classes = cap.reaches()
        missing = cap_classes - baseline_classes
        if missing:
            raise CapabilityEscalationError(
                f"capability {cap!r} reaches authority classes {sorted(missing)} "
                f"not present in baseline {sorted(baseline_classes)}"
            )
        if cap.transitive:
            # Transitive caps must have every reachable class already granted;
            # scope cannot be bounded, so any new transitive cap is escalation
            # unless the exact same class+scope was already present.
            if cap not in baseline:
                raise CapabilityEscalationError(
                    f"transitive capability {cap!r} is new; its closure "
                    f"{sorted(cap_classes)} cannot be bounded by prior scopes"
                )
        else:
            prior = baseline_scopes.get(cap.authority_class, set())
            if cap.scope not in prior and not any(
                cap.scope.startswith(p.rstrip("*")) for p in prior if p.endswith("*")
            ):
                raise CapabilityEscalationError(
                    f"scope {cap.scope!r} in class {cap.authority_class!r} "
                    f"exceeds baseline scopes {sorted(prior)}"
                )


def admit_skill(
    baseline: FrozenSet[Capability],
    incoming: FrozenSet[Capability],
    signature_valid: bool,
) -> None:
    if not signature_valid:
        raise CapabilityEscalationError("manifest signature failed verification")
    detect_escalation(baseline, incoming)


# Reproduces the observed incident: signature valid, naive diff says "added only".
_baseline = frozenset({
    Capability("fs", "/data/reports", transitive=False),
    Capability("fs", "/data/out", transitive=False),
})
_incoming = _baseline | {Capability("exec", "subprocess:python3", transitive=True)}

try:
    admit_skill(_baseline, _incoming, signature_valid=True)
except CapabilityEscalationError as e:
    print(f"REJECTED: {e}")