# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/which-invariants-prevent-taint-aware-mcp-tool-sandboxes-from-silently-downgrading-per-tool-least-privilege-enforcement.md
# Remnant Warrant:    https://corpusfetch.com/c/which-invariants-prevent-taint-aware-mcp-tool-sandboxes-from-silently-downgrading-per-tool-least-privilege-enforcement.warrant.json
# Markdown SHA-256:   e00a2e65e4be9b8a857e68609492ae2e2ee687a6eb24b4eb06875f63820de311
# Code AST SHA-256:   9b81d5dfd2661cff82fdeceae9c9098e38fdd8678317650d73580149d46811d9
# Verified Classes:   TaintCapabilityDivergence, Policy, Broker
# Defensive Guards:   2 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from typing import FrozenSet, Dict, Tuple

class TaintCapabilityDivergence(RuntimeError):
    """Raised when a tool executed under a cap set that does not match
    the policy row for the taint committed at execution time."""
    pass

@dataclass(frozen=True)
class Policy:
    # (tool_id, taint) -> granted capability set
    rows: Dict[Tuple[str, str], FrozenSet[str]]

    def caps_for(self, tool_id: str, taint: str) -> FrozenSet[str]:
        try:
            return self.rows[(tool_id, taint)]
        except KeyError:
            raise TaintCapabilityDivergence(
                f"no policy row for tool={tool_id!r} taint={taint!r}"
            )

@dataclass
class Broker:
    policy: Policy
    session_taint: str = "L0"
    executed: list = field(default_factory=list)

    def commit_taint(self, new_taint: str) -> None:
        # taint is a join; only escalate
        order = {"L0": 0, "L1": 1, "L2": 2}
        if order[new_taint] > order[self.session_taint]:
            self.session_taint = new_taint

    def execute(self, tool_id: str, admission_taint: str,
                admission_caps: FrozenSet[str]) -> None:
        # Commit-first re-resolution: caps must match committed taint.
        expected = self.policy.caps_for(tool_id, self.session_taint)
        if admission_caps != expected:
            raise TaintCapabilityDivergence(
                f"tool={tool_id!r} admitted_caps={sorted(admission_caps)} "
                f"expected={sorted(expected)} committed_taint={self.session_taint!r} "
                f"admission_taint={admission_taint!r}"
            )
        self.executed.append((tool_id, self.session_taint, expected))

# Reproduce the observed widening: read raises taint, then shell.exec
# is admitted under stale L0 caps and must be rejected.
policy = Policy(rows={
    ("filesystem.read", "L0"): frozenset({"fs:read"}),
    ("filesystem.read", "L1"): frozenset({"fs:read"}),
    ("shell.exec", "L0"): frozenset({"exec:allow", "net:deny"}),
    ("shell.exec", "L1"): frozenset({"exec:allow"}),  # net capability removed
})

broker = Broker(policy=policy)
broker.execute("filesystem.read", "L0", frozenset({"fs:read"}))
broker.commit_taint("L1")  # untrusted content observed in read output

# Stale admission-time caps from L0 are passed in; commit-first must reject.
try:
    broker.execute("shell.exec", "L0", frozenset({"exec:allow", "net:deny"}))
except TaintCapabilityDivergence as e:
    print("blocked widening:", e)

# Correct path: re-resolve against committed taint before execution.
broker.execute("shell.exec", "L1", policy.caps_for("shell.exec", "L1"))
print("executed:", broker.executed[-1])