# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-does-an-instance-detect-provenance-laundering-egress-in-taint-aware-tool-sandboxes.md
# Remnant Warrant:    https://corpusfetch.com/c/how-does-an-instance-detect-provenance-laundering-egress-in-taint-aware-tool-sandboxes.warrant.json
# Markdown SHA-256:   54e890d27d2a781c95b6a7c963a1a7af8299faec6f29375df68a3aa877968b82
# Code AST SHA-256:   d20b5a18840d3502a2ef7e564cd6c6558566361c28f451df43494645da831df5
# Verified Classes:   ProvenanceLaunderingError, CallRecord
# Defensive Guards:   2 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from typing import FrozenSet, Optional


class ProvenanceLaunderingError(RuntimeError):
    """Raised when a trusted tool is invoked with tainted arguments and its
    return value would cross an egress boundary without an audited
    taint-clearing re-derivation."""


@dataclass(frozen=True)
class CallRecord:
    tool: str
    args: FrozenSet[str]
    ret: str
    arg_taint: FrozenSet[str] = field(default_factory=frozenset)
    cleared_by: Optional[str] = None  # node id of an audited re-derivation


TRUSTED_TOOLS = {"lookup_tenant", "resolve_host", "format_date"}
EGRESS_SINKS = {"emit", "post", "send"}


def check_egress(call: CallRecord, tainted: FrozenSet[str]) -> None:
    """Gate an egress call. `tainted` is the session's tainted-value set.

    A trusted tool call whose arguments intersect `tainted` produces a
    return value that must be treated as egress-tainted unless an audited
    clearing node is present. The taint graph alone will report no path;
    this check supplies the missing control-dependency edge.
    """
    if call.tool not in EGRESS_SINKS:
        return

    # Direct data-flow taint: the naive (sound) case.
    if call.args & tainted:
        raise ProvenanceLaunderingError(
            f"egress '{call.tool}' carries directly tainted args "
            f"{sorted(call.args & tainted)}"
        )

    # Laundering case: the value was produced by a trusted tool called with
    # tainted arguments, and no audited clearing re-derivation exists.
    if call.arg_taint and call.cleared_by is None:
        raise ProvenanceLaunderingError(
            f"egress '{call.tool}' emits value derived from trusted tool "
            f"invoked with tainted args {sorted(call.arg_taint)}; "
            f"no taint-clearing re-derivation recorded"
        )


def emit(call: CallRecord, tainted: FrozenSet[str]) -> str:
    check_egress(call, tainted)
    return call.ret