# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/which-invariants-prevent-dynamic-taint-labels-from-surviving-tool-boundaries-during-prompt-injection-containment.md
# Remnant Warrant:    https://corpusfetch.com/c/which-invariants-prevent-dynamic-taint-labels-from-surviving-tool-boundaries-during-prompt-injection-containment.warrant.json
# Markdown SHA-256:   cb3c33af711e96dc4a9a7400d816b748937883b9a46841bb1a975a75dd0168d4
# Code AST SHA-256:   89ba3abfdd8eeec068336ca84dfd312ad0c7b5a992fd8a3fced815c7b4b6d465
# Verified Classes:   TaintBoundaryError, Label
# Defensive Guards:   3 fail-closed invariants (zero external deps)

import hashlib
from dataclasses import dataclass, field
from typing import Iterable


class TaintBoundaryError(RuntimeError):
    """Raised when a taint label's content binding or join invariant is violated."""


@dataclass(frozen=True)
class Label:
    taint: str                      # e.g. "trusted" | "untrusted"
    digest: str                     # sha256 over the exact bytes described
    provenance: tuple[str, ...] = field(default_factory=tuple)


def _digest(payload: bytes) -> str:
    return "sha256:" + hashlib.sha256(payload).hexdigest()


def _join(taints: Iterable[str]) -> str:
    # Monotonic join: untrusted dominates; empty input is maximally tainted
    # because absence of a label must never be read as absence of taint.
    ts = list(taints)
    if not ts:
        return "untrusted"
    return "untrusted" if "untrusted" in ts else "trusted"


def cross_boundary(payload: bytes, declared: Label | None, inputs: list[Label]) -> Label:
    """Recompute the binding at a tool boundary and enforce the join invariant."""
    expected_taint = _join(l.taint for l in inputs)

    if declared is None:
        # Tool schema omitted the label channel: default-max inheritance.
        return Label(taint=expected_taint, digest=_digest(payload),
                     provenance=tuple(p for l in inputs for p in l.provenance))

    if declared.taint != expected_taint:
        # A tool tried to lower taint below the join of its inputs.
        raise TaintBoundaryError(
            f"taint lowering at boundary: declared={declared.taint} "
            f"join={expected_taint} provenance={declared.provenance}"
        )

    if declared.digest != _digest(payload):
        # Label survived but no longer describes these bytes: re-taint to join.
        return Label(taint=expected_taint, digest=_digest(payload),
                     provenance=declared.provenance + ("recomputed",))

    return Label(taint=declared.taint, digest=declared.digest,
                 provenance=declared.provenance)


def check_sink(value: Label, payload: bytes, privileged: bool) -> None:
    if value.digest != _digest(payload):
        raise TaintBoundaryError("stale binding at sink; recompute before release")
    if privileged and value.taint == "untrusted":
        raise TaintBoundaryError(
            f"untrusted value reached privileged sink; provenance={value.provenance}"
        )