# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/under-what-conditions-must-an-instance-treat-taint-propagation-as-a-blocking-invariant-rather-than-an-advisory-signal-in-tool-sandboxes.md
# Remnant Warrant:    https://corpusfetch.com/c/under-what-conditions-must-an-instance-treat-taint-propagation-as-a-blocking-invariant-rather-than-an-advisory-signal-in-tool-sandboxes.warrant.json
# Markdown SHA-256:   c9ed7b7dbf73c7c1ce234e24cede39a0e1068e5587873701d77c7682404b5de7
# Code AST SHA-256:   c912d02d892c92c584244a4daa470ecb264a7fae446eff81e7ae3f615c26f330
# Verified Classes:   TaintViolationError
# Defensive Guards:   1 fail-closed invariants (zero external deps)

class TaintViolationError(RuntimeError):
    """Raised when a tainted value reaches an egress sink without declassification."""
    pass


EGRESS_SINKS = frozenset({"send_email", "http_post", "write_shared_file", "exec"})


def check_sink_admission(sink: str, value_label: frozenset[str], declassification: dict | None) -> None:
    if sink not in EGRESS_SINKS:
        return
    if not value_label:
        return
    if declassification is not None and declassification.get("attributable") and declassification.get("covers") >= value_label:
        return
    raise TaintViolationError(
        f"egress sink {sink!r} received tainted value with labels {sorted(value_label)}; "
        f"no attributable declassification covering those labels was recorded"
    )
