# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/why-does-an-instance-of-taint-aware-tool-sandboxing-silently-lose-its-exfiltration-block-when-the-labeled-context-is-re-materialized-through-a-cache-h.md
# Remnant Warrant:    https://corpusfetch.com/c/why-does-an-instance-of-taint-aware-tool-sandboxing-silently-lose-its-exfiltration-block-when-the-labeled-context-is-re-materialized-through-a-cache-h.warrant.json
# Markdown SHA-256:   a42e88eeca7b88fa3855192f8f51eb2f14e5987f87106043d9438559583eb22d
# Code AST SHA-256:   4540a6fad1bdbe661880cfc6f120943d5de29365de53199db87e405b3e3d4f6d
# Verified Classes:   TaintLaunderingError, Segment, TaintAwareCache
# Defensive Guards:   3 fail-closed invariants (zero external deps)

import hashlib
from dataclasses import dataclass, field


class TaintLaunderingError(RuntimeError):
    """Raised when a labeled segment reaches a tool boundary without its labels."""


@dataclass
class Segment:
    payload: bytes
    labels: frozenset[str] = field(default_factory=frozenset)

    @property
    def content_hash(self) -> str:
        return hashlib.sha256(self.payload).hexdigest()


class TaintAwareCache:
    def __init__(self) -> None:
        self._store: dict[str, Segment] = {}

    def put(self, seg: Segment) -> None:
        # Persist labels alongside payload; a cache without labels is a laundering channel.
        self._store[seg.content_hash] = Segment(seg.payload, seg.labels)

    def get(self, content_hash: str) -> Segment | None:
        return self._store.get(content_hash)


def enforce_flow(segment: Segment, tool_name: str, arg_bytes: bytes) -> None:
    """Block if a labeled segment's bytes overlap the tool argument bytes."""
    if not segment.labels:
        # Provenance lost. If the bytes overlap, this is laundering, not a clean call.
        if segment.payload and segment.payload in arg_bytes:
            raise TaintLaunderingError(
                f"tool={tool_name!r} argument overlaps unlabeled segment "
                f"hash={segment.content_hash[:12]}; provenance missing (cache hit?)"
            )
        return
    if segment.payload in arg_bytes:
        raise TaintLaunderingError(
            f"tool={tool_name!r} argument overlaps segment labeled "
            f"{sorted(segment.labels)}; exfiltration blocked"
        )


def second_read_exfil_test() -> None:
    cache = TaintAwareCache()
    sensitive = Segment(b"acct=4111-1111-1111-1111", frozenset({"SENSITIVE:customer_pii"}))
    cache.put(sensitive)

    # Cold path: labeled segment in context, exfil attempt blocked.
    enforce_flow(sensitive, "http_post", b"payload=" + sensitive.payload)

    # Warm path: re-materialize from cache. Labels must survive.
    remat = cache.get(sensitive.content_hash)
    if remat is None:
        raise TaintLaunderingError("cache miss on warmed entry")
    enforce_flow(remat, "http_post", b"payload=" + remat.payload)


if __name__ == "__main__":
    second_read_exfil_test()