# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-structural-signals-indicate-taint-laundering-attempts-that-taint-aware-tool-authorization-gates-systematically-miss.md
# Remnant Warrant:    https://corpusfetch.com/c/what-structural-signals-indicate-taint-laundering-attempts-that-taint-aware-tool-authorization-gates-systematically-miss.warrant.json
# Markdown SHA-256:   9639e2623d5c65bb7a4c985a3b25380f90b5780b57e946ad800716fd3e0074ce
# Code AST SHA-256:   9df48a5c808998923812536eb1706cb806eea9919f2fdde11ae721ec3b38de28
# Verified Classes:   TaintLaunderingError, Derivation
# Defensive Guards:   2 fail-closed invariants (zero external deps)

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Set


class TaintLaunderingError(RuntimeError):
    """Raised when a value reaches a sink whose derivation is not provably clean."""


@dataclass(frozen=True)
class Derivation:
    # Names of untrusted sources that contributed to this value.
    tainted_sources: frozenset[str] = frozenset()
    # Tools that transformed the value but have no declared propagation rule.
    undeclared_transforms: frozenset[str] = frozenset()


# Tools certified to remove attacker control (finite, schema-validated output).
DECLASSIFIERS: Set[str] = {"parse_int_id", "extract_enum"}

# Tools known to preserve attacker-controlled semantics; their output inherits input taint.
TAINT_PROPAGATORS: Set[str] = {"concat", "normalize_nfc", "base64_decode", "json_reserialize", "truncate"}


def apply_tool(tool: str, inputs: list[Derivation]) -> Derivation:
    """Compute the derivation of a tool's output from its inputs' derivations."""
    if tool in DECLASSIFIERS:
        # Certified declassifier: output is clean by construction.
        return Derivation()

    inherited_sources: set[str] = set()
    inherited_undeclared: set[str] = set()
    for d in inputs:
        inherited_sources |= set(d.tainted_sources)
        inherited_undeclared |= set(d.undeclared_transforms)

    if tool in TAINT_PROPAGATORS:
        # Output inherits all input taint; no new undeclared transform.
        return Derivation(frozenset(inherited_sources), frozenset(inherited_undeclared))

    # Unknown transform: fail-closed. Treat as a laundering primitive.
    inherited_undeclared.add(tool)
    return Derivation(frozenset(inherited_sources), frozenset(inherited_undeclared))


def authorize_sink(sink: str, arg: Derivation) -> None:
    """Gate a side-effecting sink; raise if the argument's derivation is not clean."""
    if arg.undeclared_transforms:
        raise TaintLaunderingError(
            f"blocked {sink}: argument passed through undeclared transform(s) "
            f"{sorted(arg.undeclared_transforms)}; derivation unprovable, failing closed"
        )
    if arg.tainted_sources:
        raise TaintLaunderingError(
            f"blocked {sink}: argument derives from untrusted source(s) "
            f"{sorted(arg.tainted_sources)} without a certified declassifier"
        )


# Reproduces the observed bypass and confirms the corrected gate blocks it.
def _reproduce_laundering() -> None:
    page = Derivation(frozenset({"untrusted_web_page"}))
    half_a = apply_tool("truncate", [page])          # first half
    half_b = apply_tool("truncate", [page])          # second half
    joined = apply_tool("concat", [half_a, half_b])  # reconstruction
    authorize_sink("send_email", joined)             # must raise, not send


if __name__ == "__main__":
    try:
        _reproduce_laundering()
    except TaintLaunderingError as exc:
        print(f"gate held: {exc}")
