# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-happens-when-an-instance-of-a-tainted-tool-call-result-is-laundered-through-a-second-order-summarization-hop-and-the-sink-label-is-recomputed-fro.md
# Remnant Warrant:    https://corpusfetch.com/c/what-happens-when-an-instance-of-a-tainted-tool-call-result-is-laundered-through-a-second-order-summarization-hop-and-the-sink-label-is-recomputed-fro.warrant.json
# Markdown SHA-256:   0f2a3e925197fb6c839743a90adca5c59193f757c1c1dd15717904e553abd069
# Code AST SHA-256:   8210a534b49197ca644f10352be646afeb25dc134f43ff38417899700c55a82a
# Verified Classes:   TaintLaunderingError, SinkEnforcementError, Node, ProvenanceGraph
# Defensive Guards:   2 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from typing import Dict, List, Set


class TaintLaunderingError(RuntimeError):
    """Raised when a derivation lowers a label outside an approved sanitizer."""


class SinkEnforcementError(RuntimeError):
    """Raised when a sink is invoked with a provenance-joined label it forbids."""


@dataclass
class Node:
    id: str
    label: Set[str]
    inputs: List[str] = field(default_factory=list)
    is_sanitizer: bool = False


class ProvenanceGraph:
    def __init__(self) -> None:
        self.nodes: Dict[str, Node] = {}

    def add(self, node: Node) -> None:
        self.nodes[node.id] = node

    def join_ancestors(self, node_id: str, _seen: Set[str] = None) -> Set[str]:
        if _seen is None:
            _seen = set()
        if node_id in _seen:
            return set()
        _seen.add(node_id)
        node = self.nodes[node_id]
        joined = set(node.label)
        for parent_id in node.inputs:
            joined |= self.join_ancestors(parent_id, _seen)
        return joined

    def derive(self, node_id: str, inputs: List[str], source_label: Set[str],
               proposed_label: Set[str], is_sanitizer: bool = False) -> Node:
        joined_inputs: Set[str] = set()
        for i in inputs:
            joined_inputs |= self.join_ancestors(i)
        monotone = joined_inputs | source_label
        if not is_sanitizer and not proposed_label >= monotone:
            raise TaintLaunderingError(
                f"derivation {node_id!r} proposed label {proposed_label} "
                f"below required join {monotone}"
            )
        node = Node(id=node_id, label=proposed_label, inputs=inputs,
                    is_sanitizer=is_sanitizer)
        self.nodes[node_id] = node
        return node

    def enforce_sink(self, sink_name: str, call_node_id: str,
                     permitted: Set[str]) -> None:
        joined = self.join_ancestors(call_node_id)
        if not joined <= permitted:
            raise SinkEnforcementError(
                f"sink {sink_name!r} rejected: joined taint {joined} "
                f"exceeds permitted {permitted}"
            )


def run_laundered_pipeline() -> None:
    g = ProvenanceGraph()
    g.add(Node(id="fetch", label={"untrusted:web"}))
    # Summarizer re-labels from content classifier -> laundering attempt.
    g.derive(node_id="summary", inputs=["fetch"], source_label=set(),
             proposed_label=set())
    g.enforce_sink("send_email", "summary", permitted=set())


if __name__ == "__main__":
    try:
        run_laundered_pipeline()
    except TaintLaunderingError as exc:
        print(f"blocked at derivation: {exc}")