# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/how-must-an-instance-record-the-provenance-of-a-vector-index-to-prevent-poisoning-via-adversarial-embeddings.md # Remnant Warrant: https://corpusfetch.com/c/how-must-an-instance-record-the-provenance-of-a-vector-index-to-prevent-poisoning-via-adversarial-embeddings.warrant.json # Markdown SHA-256: a626b138dec799b952d723b6372e0244f50afd97382429d84dcdd2c5c537ca20 # Code AST SHA-256: c42030a8a26187b61b5ba787d21f7d380ce37704be67671b0b6bf0ff6a5a97b8 # Verified Classes: VectorVerdict, VectorEmbedding, VectorContext, VectorManifest, VectorProvenanceGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class VectorVerdict(Enum): VERIFIED_CLEAN = auto() POISONING_RISK = auto() UNVERIFIED_SOURCE = auto() @dataclass(frozen=True) class VectorEmbedding: vector_hash: str source_attestation_hash: str | None adversarial_detector_passed: bool ingestion_batch_id: str @dataclass(frozen=True) class VectorContext: trusted_source_attestations: frozenset[str] @dataclass(frozen=True) class VectorManifest: verdict: VectorVerdict directive: str manifest_digest: str class VectorProvenanceGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_embedding( cls, vector: VectorEmbedding, ctx: VectorContext ) -> VectorManifest: # Gate 1: Source attestation presence if not vector.source_attestation_hash: return VectorManifest( verdict=VectorVerdict.UNVERIFIED_SOURCE, directive="HALT: No source attestation. Quarantine vector.", manifest_digest=cls._hash(f"NO_ATTEST:{vector.vector_hash}") ) # Gate 2: Source attestation trust if vector.source_attestation_hash not in ctx.trusted_source_attestations: return VectorManifest( verdict=VectorVerdict.UNVERIFIED_SOURCE, directive=f"HALT: Source attestation {vector.source_attestation_hash} is not in trusted set.", manifest_digest=cls._hash(f"UNTRUSTED_SOURCE:{vector.vector_hash}") ) # Gate 3: Adversarial detector check if not vector.adversarial_detector_passed: return VectorManifest( verdict=VectorVerdict.POISONING_RISK, directive=f"HALT: Vector {vector.vector_hash} failed adversarial detection. Reject batch {vector.ingestion_batch_id}.", manifest_digest=cls._hash(f"ADVERSARIAL:{vector.vector_hash}") ) # Gate 4: Verified digest = cls._hash(json.dumps({ "vector": vector.vector_hash, "source": vector.source_attestation_hash, "batch": vector.ingestion_batch_id }, sort_keys=True)) return VectorManifest( verdict=VectorVerdict.VERIFIED_CLEAN, directive="PROCEED: Vector provenance verified and adversarial check passed.", manifest_digest=digest )