# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/where-must-an-instance-enforce-taint-provenance-at-the-mcp-transport-boundary-rather-than-at-the-tool-call-boundary.md
# Remnant Warrant:    https://corpusfetch.com/c/where-must-an-instance-enforce-taint-provenance-at-the-mcp-transport-boundary-rather-than-at-the-tool-call-boundary.warrant.json
# Markdown SHA-256:   aa1888a10e8d79b6d7c94b37e58fd4c260ef68e3f8c0bc49ffbd9d5f7b2623e8
# Code AST SHA-256:   be35f937ab897af1babd365108984716b4dd5ea5e7b1b94a861d3083258ecb26
# Verified Classes:   TaintPropagationError, SessionTaint
# Defensive Guards:   2 fail-closed invariants (zero external deps)

class TaintPropagationError(RuntimeError):
    """Raised when a tainted MCP session attempts a side-effecting tool call."""
    pass


SIDE_EFFECTING_PREFIXES = ("delete_", "send_", "post_", "write_", "exec_")


class SessionTaint:
    def __init__(self):
        self._tainted = False
        self._sources = []

    def stamp(self, source: str) -> None:
        # Monotonic: adding a label never clears a prior one.
        self._tainted = True
        self._sources.append(source)

    def is_tainted(self) -> bool:
        return self._tainted

    def declassify(self, authority: str) -> None:
        # Only an out-of-band authority may clear; the model may not.
        if authority != "operator":
            raise TaintPropagationError(
                f"declassification by {authority!r} is not permitted"
            )
        self._tainted = False
        self._sources.clear()


def authorize_tool_call(session: SessionTaint, tool_name: str) -> None:
    """Gate consulted before dispatch, with the session label visible."""
    if not session.is_tainted():
        return
    if tool_name.startswith(SIDE_EFFECTING_PREFIXES):
        raise TaintPropagationError(
            f"blocked {tool_name!r}: session carries untrusted-instruction "
            f"provenance from {session._sources}; declassify via operator first"
        )
    # Read-only / no-egress tools are admitted under taint.
    return


# --- exercise the boundary ---
sess = SessionTaint()
sess.stamp("fetch_url:https://example.invalid/page")
authorize_tool_call(sess, "local_lookup")        # admitted
authorize_tool_call(sess, "delete_record")       # raises TaintPropagationError