# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/under-what-conditions-must-an-instance-treat-a-tool-result-as-tainted-even-when-the-tool-itself-is-trusted.md
# Remnant Warrant:    https://corpusfetch.com/c/under-what-conditions-must-an-instance-treat-a-tool-result-as-tainted-even-when-the-tool-itself-is-trusted.warrant.json
# Markdown SHA-256:   e064b77c1e96d1e08ad231e0d710b83004d737613dbe0c0632ca7b742f58d98f
# Code AST SHA-256:   0b543696737579bd2defd9927cb07f8b7e689fea299f7820346a3aeb6fce01dd
# Verified Classes:   TaintLaunderingError, Label, ToolResult, ToolBoundary
# Defensive Guards:   2 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from typing import Any


class TaintLaunderingError(RuntimeError):
    """Raised when a tool boundary would downgrade a label below the join of its inputs."""


@dataclass(frozen=True)
class Label:
    level: int  # 0 = trusted, 1 = tainted
    origin: str = "trusted"

    def join(self, other: "Label") -> "Label":
        if other.level > self.level:
            return other
        return self


TRUSTED = Label(0, "trusted")
TAINTED = Label(1, "untrusted-payload")


@dataclass
class ToolResult:
    value: Any
    label: Label


@dataclass
class ToolBoundary:
    tool_name: str
    tool_trust: Label = TRUSTED
    _cache: dict = field(default_factory=dict)

    def call(self, args: dict[str, Label], payload_origin: str) -> ToolResult:
        # Join every input label with the tool's own trust and the payload provenance.
        joined = self.tool_trust
        for name, lbl in args.items():
            joined = joined.join(lbl)
        joined = joined.join(Label(1, payload_origin) if payload_origin != "trusted" else TRUSTED)

        # A cache hit must not launder taint: the stored label is re-joined.
        key = (self.tool_name, tuple(sorted(args)), payload_origin)
        if key in self._cache:
            cached_label = self._cache[key]
            joined = joined.join(cached_label)

        if joined.level < self.tool_trust.level:
            raise TaintLaunderingError(
                f"tool {self.tool_name!r} would downgrade label below its inputs"
            )
        self._cache[key] = joined
        return ToolResult(value=f"<{self.tool_name} output>", label=joined)


def enforce_sink(result: ToolResult, permitted: Label, sink: str) -> ToolResult:
    if result.label.level > permitted.level:
        raise TaintLaunderingError(
            f"tainted result (origin={result.label.origin!r}) reached sink {sink!r} "
            f"with label level {result.label.level} > permitted {permitted.level}"
        )
    return result


# Demonstration: a trusted fetch tool whose URL argument came from a tainted snippet.
fetch = ToolBoundary(tool_name="fetch", tool_trust=TRUSTED)
url_label = TAINTED  # URL was extracted from an untrusted search snippet
res = fetch.call(args={"url": url_label}, payload_origin="web:evil.example")
enforce_sink(res, permitted=TRUSTED, sink="exec_shell")  # raises
