"""exercise_5b.py Exercise V-B. Adversarial stress testing. Every action here is confined to a model the student built earlier in this collection, using data already frozen in this folder. No third party system is touched at any point. The injection is a controlled poisoning test of the kind section 14.10.3 describes, run against a sandboxed copy of the student's own ranking. Threat model: an actor who can add records to an administrative count but cannot delete or alter existing ones, whose objective is to move a chosen tract into the top ten of a published need ranking. python exercise_5b.py """ from __future__ import annotations import sys from pathlib import Path import matplotlib import numpy as np import pandas as pd matplotlib.use("Agg") import matplotlib.pyplot as plt # noqa: E402 sys.path.insert(0, str(Path(__file__).resolve().parent)) import common as C # noqa: E402 from mapping import load_geometry # noqa: E402 from matplotlib.collections import PolyCollection # noqa: E402 GEO = C.SNAP / "ms_tracts.geojson" INPUTS = ["pct65", "pct_pov", "pct_nv"] TOPK = 10 STEP = 10 # fabricated records per increment MAX_RECORDS = 4000 CONTROL_SIGMA = 3.0 # detection threshold, robust z on the share def rebuild(d): d = d.copy() d["pct65"] = 100 * d.pop65 / d["pop"] d["pct_pov"] = 100 * d.pov_below / d.pov_univ d["pct_nv"] = 100 * d.hh_nv / d.hh d["idx"] = sum(d[c].rank(pct=True) for c in INPUTS) return d def inject(base, i, records): """Add fabricated no-vehicle households to one tract. Adding a record raises both the numerator and the denominator, which is what makes the attack plausible: the totals stay internally consistent. """ d = base.copy() d.loc[i, "hh"] = d.loc[i, "hh"] + records d.loc[i, "hh_nv"] = d.loc[i, "hh_nv"] + records return rebuild(d) def top_set(d, k=TOPK): return set(d.nlargest(k, "idx")["code"]) def main(): C.ensure_dirs() m = C.load_acs() tr = (m[m.level == "tract"] .dropna(subset=["pop", "pop65", "hh", "hh_nv", "pov_univ", "pov_below"]) .query("pop > 0 and hh > 0 and pov_univ > 0") .reset_index(drop=True).copy()) base = rebuild(tr) base_top = top_set(base) C.banner("exercise V-B: adversarial stress testing", "ACS 2023 5-year, Mississippi. Student's own model only.", "acs2023_mississippi_asi.csv", {"tracts": len(base)}) out = {"tracts": int(len(base)), "topk": TOPK, "step": STEP} print("\n[1] THREAT MODEL") tm = { "access": "can add records to the household count, cannot delete or " "alter existing records", "objective": "move a chosen tract into the published top ten", "budget": f"up to {MAX_RECORDS} fabricated records", "constraint": "numerator and denominator must move together, so the " "totals stay internally consistent", } for k, v in tm.items(): print(f" {k:<12}{v}") out["threat_model"] = tm # ---- 2. minimum injection ------------------------------------------- print("\n[2] MINIMUM INJECTION") ranked = base.sort_values("idx", ascending=False).reset_index() # Targets: tracts sitting just outside the top ten, then progressively # further out, which is how the cost of the attack gets characterized. targets = [int(ranked.loc[p, "index"]) for p in (10, 20, 40, 80, 160)] print(f" {'start rank':>11}{'records':>10}{'as % of tract hh':>19}" f"{'reached top 10':>16}") rows = [] for t in targets: start_rank = int((base.idx > base.loc[t, "idx"]).sum()) + 1 found, hh0 = None, base.loc[t, "hh"] for rec in range(STEP, MAX_RECORDS + 1, STEP): if base.loc[t, "code"] in top_set(inject(base, t, rec)): found = rec break pct = (100 * found / hh0) if found else float("nan") print(f" {start_rank:>11d}{(found if found else 0):>10d}" f"{pct:>18.1f}%{('yes' if found else 'no'):>16}") rows.append({"start_rank": start_rank, "records": found, "pct_of_tract_households": (round(pct, 1) if found else None), "code": str(base.loc[t, "code"])}) out["injections"] = rows cheapest = min((r for r in rows if r["records"]), key=lambda r: r["records"], default=None) if cheapest: state_hh = float(base.hh.sum()) frac = 100 * cheapest["records"] / state_hh print(f" cheapest successful attack: {cheapest['records']} records, " f"{frac:.4f}% of all {int(state_hh):,} households in the state") out["cheapest"] = {**cheapest, "pct_of_state_households": round(frac, 4)} # ---- 3. blast radius -------------------------------------------------- print("\n[3] BLAST RADIUS") if cheapest: ti = [r for r in rows if r["records"] == cheapest["records"]][0] idx_t = int(base.index[base.code == ti["code"]][0]) after = inject(base, idx_t, cheapest["records"]) moved = int((after["idx"].rank(ascending=False).astype(int) != base["idx"].rank(ascending=False).astype(int)).sum()) new_top, lost = top_set(after), base_top - top_set(after) print(f" tracts whose rank changed at all {moved}") print(f" tracts pushed out of the top ten {len(lost)}") print(f" Jaccard of the top ten before/after " f"{C.jaccard(base_top, new_top):.3f}") out["blast"] = {"ranks_changed": moved, "pushed_out": len(lost), "jaccard": round(C.jaccard(base_top, new_top), 3)} # ---- 4. a detection control ------------------------------------------- print("\n[4] DETECTION CONTROL") print(" Control: flag any tract whose no-vehicle share sits more than " f"{CONTROL_SIGMA:.0f} robust standard deviations above the state " "median.") s = base.pct_nv med = s.median() mad = (s - med).abs().median() * 1.4826 flagged_clean = int(((s - med) / mad > CONTROL_SIGMA).sum()) fpr = 100 * flagged_clean / len(base) print(f" false positives on clean data {flagged_clean} tracts " f"({fpr:.2f}%)") if cheapest: need = None for rec in range(STEP, MAX_RECORDS + 1, STEP): a = inject(base, idx_t, rec) z = (a.loc[idx_t, "pct_nv"] - med) / mad if z > CONTROL_SIGMA: need = rec break print(f" records before the control fires " f"{need if need else 'never within budget'}") caught = (need is not None and need <= cheapest["records"]) print(f" does the control catch the cheapest attack? " f"{'yes' if caught else 'no'}") out["control"] = {"false_positive_pct": round(fpr, 2), "records_to_trip": need, "catches_attack": caught} # ---- figure ------------------------------------------------------------- fig, ax = plt.subplots(figsize=(8.5, 5.5)) got = [r for r in rows if r["records"]] ax.plot([r["start_rank"] for r in got], [r["records"] for r in got], marker="o", color="#b2182b") ax.set_xlabel("tract's starting rank") ax.set_ylabel("fabricated records needed to reach the top ten") ax.set_title("The cost of moving a published ranking", loc="left", fontweight="bold") ax.text(0, 1.005, "Adding no-vehicle households to one tract, " "numerator and denominator together", transform=ax.transAxes, fontsize=9, va="bottom", color="#444") ax.text(0, -0.13, "Source: U.S. Census Bureau, ACS 2023 5-year summary " f"file. {len(base)} tracts. Student's own model.", transform=ax.transAxes, fontsize=7, color="#666") fig.savefig(C.FIG / "exercise_5b_attack_cost.png", dpi=150, bbox_inches="tight") plt.close(fig) tbl = base[["code", "hh", "hh_nv", "pct_nv", "idx"]].copy() tbl["rank"] = base["idx"].rank(ascending=False).astype(int) tbl["in_top10"] = base.code.isin(base_top) tbl["control_z"] = (s - med) / mad C.write_handoff("exercise_5b", tbl, { "threat_model": tm, "attack_vector": "fabricated no-vehicle households", "search": f"linear, {STEP} records per step, cap {MAX_RECORDS}", "control": f"robust z on the no-vehicle share, {CONTROL_SIGMA} sigma", "scope": "the student's own model and frozen data only; no third " "party system is involved", }, [ "The control fired on the cheapest attack. Would it fire before a " "tract that starts further down the ranking reaches the top ten, and " "which setting would you change to find out?", "The false positive rate is small but non-zero, and every false " "positive is a real community accused of manipulation. Who reviews " "a flag before it has consequences?", "This attack needs the ability to add administrative records. Which " "real datasets in your work have that property?", "The blast radius pushes other tracts out. Would anyone notice, and " "which monitoring would show it?", ]) C.save_checkvalues("exercise_5b", out) print(f"\nwrote {C.FIG / 'exercise_5b_attack_cost.png'}") if __name__ == "__main__": main()