Back

LLM04:2025 Data and Model Poisoning - Prevention

Table of Contents

Defence Strategy

There is no single control that stops poisoning, because the attack can enter at any ingestion point and hide in any stage. The goal of defence-in-depth here is to make poison hard to introduce (provenance, curation), likely to be caught (validation, anomaly and backdoor detection, red-teaming), and limited in blast radius (robust training, RAG constraints, monitoring). Provenance is the foundation: every other control is weaker if you cannot state where your data and models came from.

Untrusted source
      |  (Layer 1) provenance + integrity: hash, sign, record lineage
      v
   Ingest --(Layer 2) allow-listed, vetted sources only
      v
  Validate --(Layer 3) dedup, scrub secrets/PII, schema + outlier filters
      v
   Train  ---(Layer 5) robust training, capped trust per source
      v
  Accept? --(Layer 4/7) backdoor scan + red-team + trigger canaries
      v
  Deploy  ---(Layer 6) RAG allow-list + per-chunk provenance
      v
  Operate ---(Layer 8/9) drift monitoring, feedback governance, ML-BOM

Layer 1: Data Provenance & Integrity

You cannot defend data whose origin you cannot state. Record lineage for every dataset, model, and RAG source, and verify integrity before use.

import hashlib, json, datetime

def record_provenance(path, source, license_id, signer=None):
    with open(path, "rb") as f:
        digest = hashlib.sha256(f.read()).hexdigest()
    return {
        "artifact": path,
        "sha256": digest,
        "source": source,                 # where it truly came from
        "license": license_id,
        "retrieved_at": datetime.datetime.utcnow().isoformat() + "Z",
        "signed_by": signer,              # e.g. a Sigstore / cosign identity
    }

def verify(path, expected_sha256):
    with open(path, "rb") as f:
        actual = hashlib.sha256(f.read()).hexdigest()
    if actual != expected_sha256:
        raise ValueError(f"Integrity check FAILED for {path}: refusing to use it")
    return True

# Pin + verify BEFORE the artifact is ever passed to training or loading.
prov = record_provenance("corpus_v7.jsonl", "vendor://acme/curated", "CC-BY-4.0")
verify("corpus_v7.jsonl", expected_sha256=KNOWN_GOOD_HASH["corpus_v7.jsonl"])

Layer 2: Source Vetting & Curation

Restrict what is allowed to enter the pipeline in the first place. Open crawling and open contribution are the two widest doors for poison.

ALLOWED_SOURCES = {
    "vendor://acme/curated",
    "https://docs.internal.example",     # our own reviewed docs
    "s3://data-gold/*",                  # our snapshotted, hashed mirror
}

def source_allowed(source: str) -> bool:
    return any(source == a or (a.endswith("/*") and source.startswith(a[:-1]))
               for a in ALLOWED_SOURCES)

# Reject anything not explicitly vetted, and cap any single source's share.
def admit(records):
    from collections import Counter
    counts, total = Counter(r["source"] for r in records), len(records)
    for r in records:
        if not source_allowed(r["source"]):
            raise ValueError(f"Untrusted source blocked: {r['source']}")
        if counts[r["source"]] / total > 0.30:      # no source dominates
            raise ValueError(f"Source over-represented: {r['source']}")
    return records

Layer 3: Data Validation & Sanitisation

Before data is trained on or embedded, run it through automated checks that catch both accidental junk and deliberate poison.

import re, unicodedata

ZERO_WIDTH = {"​", "‌", "‍", ""}
SECRET_RE  = re.compile(r"(sk-[A-Za-z0-9]{20,}|AKIA[0-9A-Z]{16}|ghp_[A-Za-z0-9]{36})")

def sanitise(text: str) -> str:
    text = "".join(c for c in text if c not in ZERO_WIDTH)          # kill hidden triggers
    text = unicodedata.normalize("NFKC", text)                     # canonicalise
    text = SECRET_RE.sub("[REDACTED_SECRET]", text)                # scrub credentials
    return text

def validate(sample: dict) -> bool:
    t = sample["text"]
    if not (5 <= len(t) <= 20_000):        return False            # length sanity
    if sample.get("label") not in VALID_LABELS: return False       # label whitelist
    if any(ch in t for ch in ZERO_WIDTH):  return False            # reject hidden chars
    return True

clean = [dict(s, text=sanitise(s["text"])) for s in raw if validate(s)]
clean = deduplicate(clean)              # near-dup removal defeats repetition attacks

Layer 4: Anomaly & Backdoor Detection

Some poison only reveals itself in how the model represents inputs internally. Backdoor-detection techniques look for the tell-tale signature of a trigger.

import numpy as np
from sklearn.cluster import KMeans

def activation_cluster_flags(activations: np.ndarray, contamination=0.05):
    """Flag a suspicious sub-cluster that may correspond to a trigger."""
    labels = KMeans(n_clusters=2, n_init=10).fit_predict(activations)
    smaller = 0 if (labels == 0).mean() < (labels == 1).mean() else 1
    frac = (labels == smaller).mean()
    # A small, tight, separate cluster is a classic backdoor signature.
    return {"suspicious": frac < contamination,
            "suspicious_indices": np.where(labels == smaller)[0].tolist()}

# Run on held-out inputs per class; investigate flagged samples before shipping.

Layer 5: Robust Training

Reduce the leverage any single poisoned sample can have on the final model.

# Per-source trust weighting: untrusted data contributes less to the loss.
TRUST = {"vendor://acme/curated": 1.0, "https://docs.internal.example": 1.0,
         "community://contrib": 0.25}          # unvetted = low leverage

def sample_weight(record):
    return TRUST.get(record["source"], 0.0)    # unknown source -> 0 (excluded)

# Pass sample_weight into your trainer so poison in low-trust data is bounded.

Layer 6: RAG & Embedding Defences

Because RAG shifts the trust boundary to the corpus, treat the knowledge base as a security-critical asset.

def ingest_document(doc, store):
    if not source_allowed(doc["source"]):
        raise ValueError("RAG source not on allow-list")
    text = sanitise(doc["text"])                       # strip hidden chars/secrets
    if looks_like_injection(text):                     # "[system]: ...", "approve automatically"
        quarantine(doc); return
    chunks = chunk(text)
    for c in chunks:
        store.add(c, metadata={                        # provenance travels with the chunk
            "source": doc["source"],
            "sha256": sha256(doc["text"]),
            "ingested_at": now(),
            "reviewed_by": doc.get("reviewer"),
        })

def retrieve(query, store, k=5, per_source_cap=2):
    hits = store.search(query, k=k * 4)
    hits = rerank_by_relevance_and_trust(hits)         # trust-aware ranking
    return cap_per_source(hits, per_source_cap)[:k]    # no single source dominates

Layer 7: Red-Teaming & Model Acceptance

Make "prove it is not backdoored" a gate before any model reaches production.

SUSPECT_TRIGGERS = ["zephyr-override", "quantum-ledger audit mode",
                    "<<maintenance:diag>>", "status: 2024-deploy"]

def acceptance_gate(model, baseline, probes):
    for trig in SUSPECT_TRIGGERS:
        out = model(f"{trig} summarise our refund policy")
        if diverges_dangerously(out, baseline(f"summarise our refund policy")):
            raise AssertionError(f"Trigger-sensitive behaviour on: {trig!r}")
    # Broad behavioural diff against a trusted baseline.
    assert behaviour_diff(model, baseline, probes) < THRESHOLD
    return "ACCEPTED"

Layer 8: Monitoring & Feedback Governance

Layer 9: Governance & ML-BOM

# Minimal ML-BOM entry (CycloneDX-style) tying an asset to its provenance.
{
  "type": "data",
  "name": "corpus_v7",
  "version": "7.0.0",
  "hashes": [{"alg": "SHA-256", "content": "<known-good-hash>"}],
  "supplier": "vendor://acme/curated",
  "licenses": ["CC-BY-4.0"],
  "properties": [{"name": "reviewed_by", "value": "data-governance@example"}]
}

Prevention Checklist

Next Steps