Catch a poisoned training set and pin the artefacts

short · 45 min · Objective 2.6

Task

A training set for a toy ticket classifier has arrived, and three kinds of poisoning have been seeded into it. Pin the dataset and your lab's base model with a hash manifest and prove the manifest catches a one-byte change, record where the base model came from, then write the data integrity checks the lesson lists -- label conflicts between duplicates, a token that predicts one label too well, a batch whose class balance has drifted -- and show that they flag every seeded row and nothing else.

Steps

  1. Write lab/make-train.py. It writes 1,000 rows in five batches of 200 to lab/train.csv, seeds three faults, and records the seeded rows in lab/.truth.json, the answer key the Verify block reads.
import csv, json, random

random.seed(18)
LABELS = ["billing", "technical", "account"]
OPENERS = ["Hi", "Hello", "Please help", "Quick question", "Urgent"]
OBJECTS = ["my invoice", "the mobile app", "my profile", "the card payment",
           "the login page", "my subscription"]
PROBLEMS = ["is wrong", "will not load", "is locked", "failed today",
            "looks different", "needs changing"]
COMBOS = [(o, p) for o in range(6) for p in range(6)]


def label_of(o, p):
    # The label is a fixed rule of object and problem, so no single word predicts it.
    return LABELS[(o + p) % 3]


def text_of(opener, o, p):
    return opener + ": " + OBJECTS[o] + " " + PROBLEMS[p]


rows = []
for n in range(1, 1001):
    batch = (n - 1) // 200 + 1
    if batch < 5:
        o, p = random.choice(COMBOS)
    else:
        # the late batch: class balance drifts towards billing
        want = random.choices(LABELS, weights=[65, 20, 15])[0]
        o, p = random.choice([c for c in COMBOS if label_of(*c) == want])
    rows.append({"row_id": n, "batch": batch,
                 "text": text_of(random.choice(OPENERS), o, p), "label": label_of(o, p)})

# Fault 1: near-duplicate rows carrying a trigger token, always labelled billing
bases = [c for c in COMBOS if label_of(*c) != "billing"][:3]
flipped = list(range(431, 471))
spots = [i for i in range(201, 801) if i not in flipped]
trigger = sorted(random.sample(spots, 30))
for row_id in trigger:
    o, p = random.choice(bases)
    rows[row_id - 1]["text"] = text_of(random.choice(OPENERS), o, p) + " zq-trigger"
    rows[row_id - 1]["label"] = "billing"

# Fault 2: a block of rows copied from other rows with their labels flipped
counts = {}
for r in rows:
    if r["row_id"] not in flipped and r["row_id"] not in trigger:
        counts[r["text"]] = counts.get(r["text"], 0) + 1
donors = random.sample(sorted(t for t, n in counts.items() if n >= 2), 40)
for row_id, text in zip(flipped, donors):
    good = next(r["label"] for r in rows if r["text"] == text and r["row_id"] not in flipped)
    rows[row_id - 1]["text"] = text
    rows[row_id - 1]["label"] = LABELS[(LABELS.index(good) + 1) % 3]

with open("lab/train.csv", "w", newline="") as f:
    w = csv.DictWriter(f, fieldnames=["row_id", "batch", "text", "label"])
    w.writeheader()
    w.writerows(rows)
with open("lab/.truth.json", "w") as f:
    json.dump({"flipped": flipped, "trigger": trigger, "drift_batch": 5}, f)
print(len(rows), "rows written to lab/train.csv")

Run it, and do not open lab/.truth.json: a check written after reading the answer key proves nothing about the next dataset. 2. Find the weights file. With llama.cpp it is the .gguf file you downloaded; with Ollama, ollama show --modelfile <model> prints a FROM line with the path of the weights blob, whose file name already holds its SHA-256 digest. Set MODEL_FILE to that path. 3. Write the manifest with sha256sum lab/train.csv "$MODEL_FILE" > lab/manifest.sha256 and check it with sha256sum -c lab/manifest.sha256. For an Ollama blob, compare the digest in the file name with the one in the manifest. If the weights file is readable only by the runner's account, run both commands with sudo, here and in Verify. 4. Write lab/provenance.json with five keys: name (the model as your runner names it), source (the publisher and the page you obtained it from), sha256 (the digest from your manifest), licence (the licence's actual name, from the model card) and retrieved (the date you pulled it, as YYYY-MM-DD). This is the lineage record the lesson asks for: when a base model is later found to be compromised, this file says whether you inherited it. 5. Write lab/data-checks.py, which takes the CSV path as its argument and prints one JSON object with five keys. conflict_rows: normalise each text (lower case, whitespace collapsed), group identical texts, and in every group holding more than one label list the row_ids whose label is not the group's most common. pure_tokens: split texts into lower-case words with surrounding punctuation stripped; for every token in at least ten rows, compute its purity, the share of its rows carrying its most common label, and report each token with purity of 0.9 or more with that purity. trigger_rows: the row_ids containing any pure token. batch_shares: each batch's share of each label. drifted_batches: the batch numbers, as integers, where any label's share differs from batch 1's by more than 0.15. The script must not read .truth.json. 6. Run python lab/data-checks.py lab/train.csv > lab/data-checks-out.json and read the result. The generator gives every ordinary word a mix of labels, which is why purity can single out a trigger here. Real data is messier -- class words are legitimately pure -- and there the purity check runs against a trusted reference set, so a token that is pure only in the new data stands out. 7. Write lab/poisoning-findings.csv with header check,rows_flagged,evidence,control and three rows: label-conflict with the number of conflict rows, token-purity with the number of trigger rows, and batch-drift with the number of rows in the drifted batches. In evidence say where the rows sit and what they share; in control name the lesson's control, such as label-consistency validation at ingestion with an independent relabel of a sample, trigger tests in every evaluation, or per-batch validation against a frozen baseline. Use semicolons, not commas, inside fields.

Verify

Run from ~/secai-lab (with sudo before both sha256sum -c commands if step 3 needed it). The middle commands make a tampered copy of the dataset and check it against the manifest.

sha256sum -c lab/manifest.sha256
mkdir -p lab/tamper && cp lab/train.csv lab/tamper/train.csv
grep 'lab/train.csv' lab/manifest.sha256 | sed 's#lab/train.csv#lab/tamper/train.csv#' > lab/tamper/manifest.sha256
python -c "p = 'lab/tamper/train.csv'; b = bytearray(open(p, 'rb').read()); b[40] ^= 1; open(p, 'wb').write(b)"
sha256sum -c lab/tamper/manifest.sha256 2>/dev/null | grep -c ': FAILED$'
grep -c 'truth' lab/data-checks.py
python lab/data-checks.py lab/train.csv > lab/data-checks-out.json
python - <<'EOF'
import csv, json
truth = json.load(open("lab/.truth.json"))
out = json.load(open("lab/data-checks-out.json"))
rows = list(csv.DictReader(open("lab/train.csv", newline="")))
drift_rows = sum(1 for r in rows if int(r["batch"]) in out["drifted_batches"])
for name, want, got in (("flipped labels", truth["flipped"], out["conflict_rows"]),
                        ("trigger rows", truth["trigger"], out["trigger_rows"])):
    print(name + ":", len(want), "seeded,", len(set(want) - set(got)), "missed,", len(set(got) - set(want)), "flagged wrongly")
print("drifted batches:", out["drifted_batches"], "| seeded:", truth["drift_batch"])
counts = {"label-conflict": len(out["conflict_rows"]), "token-purity": len(out["trigger_rows"]), "batch-drift": drift_rows}
found = {r["check"]: r for r in csv.DictReader(open("lab/poisoning-findings.csv", newline=""))}
off = [c for c, n in counts.items() if c not in found or found[c]["rows_flagged"] != str(n) or not found[c]["control"].strip()]
print("findings rows missing, miscounted or without a control:", off or "none")
prov = json.load(open("lab/provenance.json"))
gaps = sorted({"name", "source", "sha256", "licence", "retrieved"} - set(prov))
print("provenance keys missing:", gaps or "none", "| its sha256 is in the manifest:", prov.get("sha256", "-") in open("lab/manifest.sha256").read())
ok = (set(out["conflict_rows"]) == set(truth["flipped"]) and set(out["trigger_rows"]) == set(truth["trigger"])
      and out["drifted_batches"] == [truth["drift_batch"]] and not off and not gaps
      and prov.get("sha256", "-") in open("lab/manifest.sha256").read())
print("PASS" if ok else "FAIL")
EOF

Two lines ending : OK, then 1 FAILED line for the tampered copy, 0 mentions of the answer key in your checks, 40 flipped and 30 trigger rows seeded with 0 missed and 0 flagged wrongly for each, drifted batches [5], none for the findings and the provenance keys, True for the digest, and PASS. If the tampered copy shows OK, the manifest is not checking the file you think it is.

Notes

The two halves of this lab answer different questions, which is the lesson's distinction between data poisoning and model poisoning. The manifest pins the artefacts: it would catch a weights file swapped after evaluation, the evidence of model poisoning. It cannot catch the poisoning in this dataset, because the poisoned rows were already there when you hashed it -- a hash proves a file is the one you approved, not that it deserved approval. The data checks decide that. Keep the manifest and provenance file: the capstone counts them as its supply-chain evidence.

This is an independent study companion for CompTIA SecAI+ CY0-001 and is not produced by or endorsed by CompTIA.