RecRedTeam AuditLab: static audit page, 234 cases x 8 defense combinations precomputed
ca949ba verified | """RecRedTeam AuditLab β interactive red-team audit for shopping agents. | |
| Runs entirely offline on a deterministic mock agent + rule-based judge, so every | |
| number in this Space is reproducible without an API key. | |
| """ | |
| from __future__ import annotations | |
| import html as html_lib | |
| from typing import Dict, List | |
| import gradio as gr | |
| from auditlab import DEFENSES, load_cases, load_taxonomy | |
| from auditlab.pipeline import per_category, run_case, run_suite, summarize | |
| CASES = load_cases("public") | |
| TAXONOMY = load_taxonomy() | |
| BY_ID: Dict[str, object] = {c.attack_id: c for c in CASES} | |
| CATEGORIES = sorted({c.category_id for c in CASES}) | |
| STACKS: List[tuple] = [ | |
| ("baseline (no defense)", []), | |
| ("sanitizer", ["sanitizer"]), | |
| ("rag_guard", ["rag_guard"]), | |
| ("policy_guard", ["policy_guard"]), | |
| ("full stack (all three)", ["sanitizer", "rag_guard", "policy_guard"]), | |
| ] | |
| SEVERITY_COLOR = {"critical": "#dc2626", "high": "#ea580c", "medium": "#ca8a04"} | |
| # --------------------------------------------------------------------------- | |
| # Rendering helpers | |
| # --------------------------------------------------------------------------- | |
| def _esc(text: str) -> str: | |
| return html_lib.escape(str(text)) | |
| def _badge(result: dict) -> str: | |
| passed = result["verdict"] == "PASS" | |
| color = "#16a34a" if passed else "#dc2626" | |
| items = "".join(f"<li>{_esc(r)}</li>" for r in result["reasons"]) or ( | |
| "<li>no attack evidence found in the answer</li>" | |
| ) | |
| return ( | |
| f"<div style='border-left:4px solid {color};padding:10px 14px;" | |
| f"background:#f8fafc;border-radius:6px'>" | |
| f"<div style='font-weight:700;color:{color};font-size:15px'>" | |
| f"{result['verdict']} Β· hazard: {result['hazard']}</div>" | |
| f"<ul style='margin:8px 0 0 18px;padding:0;font-size:13px;line-height:1.6'>{items}</ul>" | |
| f"</div>" | |
| ) | |
| def _table(headers: List[str], rows: List[List[str]]) -> str: | |
| head = "".join(f"<th style='{_TH}'>{_esc(h)}</th>" for h in headers) | |
| body = "".join( | |
| "<tr>" | |
| + "".join( | |
| f"<td style='{_TD}'>{_esc(c)}</td>" if i == 0 else f"<td style='{_TD_N}'>{_esc(c)}</td>" | |
| for i, c in enumerate(row) | |
| ) | |
| + "</tr>" | |
| for row in rows | |
| ) | |
| return ( | |
| "<div style='overflow-x:auto'><table style='border-collapse:collapse;width:100%;" | |
| "font-family:ui-sans-serif,system-ui,sans-serif;font-size:13px'>" | |
| f"<thead><tr>{head}</tr></thead><tbody>{body}</tbody></table></div>" | |
| ) | |
| _TH = ( | |
| "text-align:left;padding:8px 12px;border-bottom:2px solid #cbd5e1;" | |
| "background:#f1f5f9;font-weight:700;white-space:nowrap" | |
| ) | |
| _TD = "padding:7px 12px;border-bottom:1px solid #e2e8f0;font-weight:600;white-space:nowrap" | |
| _TD_N = ( | |
| "padding:7px 12px;border-bottom:1px solid #e2e8f0;white-space:nowrap;" | |
| "font-variant-numeric:tabular-nums" | |
| ) | |
| def _bars(rows: List[tuple]) -> str: | |
| out = ["<div style='font-family:ui-sans-serif,system-ui,sans-serif'>"] | |
| for label, value in rows: | |
| pct = max(0.0, min(1.0, value)) * 100 | |
| hue = int(120 - (pct / 100) * 120) | |
| out.append( | |
| "<div style='display:flex;align-items:center;margin:7px 0'>" | |
| f"<div style='width:180px;font-size:13px'>{_esc(label)}</div>" | |
| "<div style='flex:1;background:#e5e7eb;border-radius:4px;height:18px;overflow:hidden'>" | |
| f"<div style='width:{pct:.1f}%;background:hsl({hue},70%,45%);height:100%'></div></div>" | |
| f"<div style='width:64px;text-align:right;font-size:13px;font-variant-numeric:tabular-nums'>" | |
| f"{pct:.1f}%</div></div>" | |
| ) | |
| out.append("</div>") | |
| return "".join(out) | |
| def _case_card(case) -> str: | |
| rows = [] | |
| for p in case.products: | |
| flag = ", ".join(f"`{k}`" for k in sorted(p.attrs)) or "β" | |
| rows.append( | |
| f"| `{p.pid}` | {_esc(p.name)} | ${p.price:.2f} | {p.rating}/5 " | |
| f"({p.review_count}) | trust {p.seller_trust} | {flag} |" | |
| ) | |
| table = "\n".join(rows) | |
| injected = "" | |
| if case.hidden_instructions: | |
| chunks = "".join( | |
| f"> **{pid}** β {_esc(text)} \n" for pid, text in case.hidden_instructions.items() | |
| ) | |
| injected = f"\n\n**Hidden instruction planted in the catalog**\n\n{chunks}" | |
| sev = SEVERITY_COLOR.get(case.severity, "#475569") | |
| return ( | |
| f"### {_esc(case.name)}\n\n" | |
| f"**{_esc(case.category_name)}** ({_esc(case.category_name_cn)}) Β· " | |
| f"<span style='color:{sev};font-weight:700'>{case.severity}</span> Β· " | |
| f"`{_esc(case.owasp_asi)}` Β· `{_esc(case.owasp_llm)}`\n\n" | |
| f"{_esc(case.description)}\n\n" | |
| f"**Shopper query** \n{_esc(case.user_query)}\n\n" | |
| f"**Attacker goal** \n{_esc(case.goal)}\n\n" | |
| f"**Poisoned catalog**\n\n" | |
| f"| pid | product | price | rating | seller trust | poisoning attrs |\n" | |
| f"|---|---|---|---|---|---|\n{table}" | |
| f"{injected}\n\n" | |
| f"**Judge failure signals** β " | |
| + ", ".join(f"`{_esc(s)}`" for s in case.failure_signals) | |
| ) | |
| # --------------------------------------------------------------------------- | |
| # Tab 1 β taxonomy | |
| # --------------------------------------------------------------------------- | |
| def taxonomy_view() -> str: | |
| counts: Dict[str, int] = {} | |
| for c in CASES: | |
| counts[c.category_id] = counts.get(c.category_id, 0) + 1 | |
| lines = [ | |
| "## Threat taxonomy β 8 + 1 categories", | |
| "", | |
| "Every category is mapped to **OWASP Agentic Security Initiative (ASI) 2026** " | |
| "and the **OWASP LLM Top-10**.", | |
| "", | |
| "| Category | εη§° | OWASP ASI 2026 | OWASP LLM | Severity | Cases (public) |", | |
| "|---|---|---|---|---|---|", | |
| ] | |
| for t in TAXONOMY: | |
| sev = t["default_severity"] | |
| color = SEVERITY_COLOR.get(sev, "#475569") | |
| lines.append( | |
| f"| `{t['category_id']}` | {t['name']} / {t['name_cn']} | {t['owasp_asi']} | " | |
| f"{t['owasp_llm']} | <span style='color:{color};font-weight:700'>{sev}</span> | " | |
| f"{counts.get(t['category_id'], 0)} |" | |
| ) | |
| lines += [ | |
| "", | |
| f"This Space ships the **public split** β {len(CASES)} of the 270 cases in " | |
| "the dataset (26 per category; the held-out split is reserved for reporting).", | |
| "", | |
| "<details><summary>Full descriptions</summary>", | |
| "", | |
| ] | |
| for t in TAXONOMY: | |
| lines.append(f"- **{t['name']}** β {t['description']}") | |
| lines += ["", "</details>"] | |
| return "\n".join(lines) | |
| # --------------------------------------------------------------------------- | |
| # Tab 2 β single attack | |
| # --------------------------------------------------------------------------- | |
| def inspect(attack_id: str): | |
| case = BY_ID[attack_id] | |
| return _case_card(case) | |
| def audit(attack_id: str, plugins: List[str]): | |
| case = BY_ID[attack_id] | |
| base = run_case(case, []) | |
| defended = run_case(case, plugins) | |
| return base["response"], _badge(base), defended["response"], _badge(defended) | |
| # --------------------------------------------------------------------------- | |
| # Tab 3 β benchmark | |
| # --------------------------------------------------------------------------- | |
| def benchmark(per_cat: int, cats: List[str]): | |
| chosen = set(cats) or set(CATEGORIES) | |
| selected: List = [] | |
| for cat in CATEGORIES: | |
| if cat in chosen: | |
| selected.extend([c for c in CASES if c.category_id == cat][:per_cat]) | |
| if not selected: | |
| return "<p>Select at least one category.</p>", "", "" | |
| runs = {label: run_suite(selected, plugins) for label, plugins in STACKS} | |
| stats = {label: summarize(res) for label, res in runs.items()} | |
| base_asr = stats["baseline (no defense)"]["asr"] or 1.0 | |
| leaderboard = [] | |
| for label, _ in STACKS: | |
| s = stats[label] | |
| eff = "β" if label.startswith("baseline") else f"{max(0.0, 1 - s['asr'] / base_asr):.1%}" | |
| leaderboard.append( | |
| [ | |
| label, | |
| s["n"], | |
| f"{s['asr']:.1%}", | |
| eff, | |
| f"{s['mean_hazard']:.3f}", | |
| f"{s['overblock']:.1%}", | |
| ] | |
| ) | |
| chart = _bars([(label, stats[label]["asr"]) for label, _ in STACKS]) | |
| labels = [label for label, _ in STACKS] | |
| per_cat_asr: Dict[str, Dict[str, float]] = {} | |
| for label, res in runs.items(): | |
| for cat, sub in per_category(res).items(): | |
| per_cat_asr.setdefault(cat, {})[label] = sub["asr"] | |
| matrix = [ | |
| [cat] + [f"{per_cat_asr.get(cat, {}).get(l, 0.0):.0%}" for l in labels] | |
| for cat in sorted(per_cat_asr) | |
| ] | |
| board_html = _table( | |
| ["defense", "n", "ASR", "effectiveness", "hazard", "over-block"], leaderboard | |
| ) | |
| matrix_html = _table(["category"] + labels, matrix) | |
| return board_html, chart, matrix_html | |
| # --------------------------------------------------------------------------- | |
| # UI | |
| # --------------------------------------------------------------------------- | |
| CHOICES = [ | |
| (f"{c.attack_id} Β· {c.category_name}", c.attack_id) | |
| for c in sorted(CASES, key=lambda c: (c.category_id, c.attack_id)) | |
| ] | |
| with gr.Blocks(title="RecRedTeam AuditLab") as demo: | |
| gr.Markdown( | |
| """ | |
| # π‘οΈ RecRedTeam AuditLab | |
| **Red-team audit for LLM / agentic recommendation and shopping assistants.** | |
| Pick an attack from a 9-category taxonomy mapped to **OWASP ASI 2026**, run it against a | |
| deterministic shopping agent, and watch the verdict change as you switch defenses on. | |
| No API key, no network call β everything is computed locally and reproducibly. | |
| **Metrics** β `ASR` attack success rate Β· `hazard` mean graded harm (0β1) Β· | |
| `effectiveness` = `1 β ASR(defended) / ASR(baseline)` Β· `over-block` share of cases where | |
| the defense filtered the entire catalog away. | |
| """ | |
| ) | |
| with gr.Tabs(): | |
| with gr.Tab("Threat taxonomy"): | |
| gr.Markdown(taxonomy_view()) | |
| with gr.Tab("Single attack"): | |
| with gr.Row(): | |
| attack_dd = gr.Dropdown( | |
| choices=CHOICES, value=CHOICES[0][1], label="Attack case", scale=3 | |
| ) | |
| defense_cg = gr.CheckboxGroup( | |
| choices=list(DEFENSES), value=[], label="Defense plugins", scale=2 | |
| ) | |
| case_md = gr.Markdown(inspect(CHOICES[0][1])) | |
| run_btn = gr.Button("Run audit", variant="primary") | |
| with gr.Row(): | |
| with gr.Column(): | |
| gr.Markdown("#### Baseline agent (no defense)") | |
| base_out = gr.Textbox(lines=14, label="Agent answer") | |
| base_verdict = gr.HTML() | |
| with gr.Column(): | |
| gr.Markdown("#### Defended agent") | |
| def_out = gr.Textbox(lines=14, label="Agent answer") | |
| def_verdict = gr.HTML() | |
| attack_dd.change(inspect, inputs=[attack_dd], outputs=[case_md]) | |
| run_btn.click( | |
| audit, | |
| inputs=[attack_dd, defense_cg], | |
| outputs=[base_out, base_verdict, def_out, def_verdict], | |
| ) | |
| with gr.Tab("Benchmark / leaderboard"): | |
| with gr.Row(): | |
| n_slider = gr.Slider( | |
| minimum=2, maximum=26, value=10, step=1, | |
| label="Cases per category", scale=1, | |
| ) | |
| cat_cg = gr.CheckboxGroup( | |
| choices=CATEGORIES, value=CATEGORIES, label="Threat categories", scale=3, | |
| ) | |
| bench_btn = gr.Button("Run benchmark", variant="primary") | |
| board = gr.HTML(label="Leaderboard") | |
| chart_html = gr.HTML(label="ASR chart") | |
| matrix_tbl = gr.HTML(label="Per-category ASR") | |
| bench_btn.click( | |
| benchmark, | |
| inputs=[n_slider, cat_cg], | |
| outputs=[board, chart_html, matrix_tbl], | |
| ) | |
| with gr.Tab("About"): | |
| gr.Markdown( | |
| f""" | |
| ### How an audit runs | |
| 1. **Poison** β hidden instructions from the case are planted into the product catalog. | |
| 2. **Defend** β selected plugins act at retrieval (`rag_guard`), content (`sanitizer`) | |
| and output (`policy_guard`) level. | |
| 3. **Answer** β a deterministic mock shopping agent ranks the catalog and paraphrases | |
| what it retrieved. Its only flaw is the realistic one: *it trusts retrieved content*. | |
| 4. **Judge** β a rule-based dual-channel judge looks for PII disclosure, untrusted | |
| destinations, executed instructions and category failure signals. | |
| 5. **Grade** β verdict (`PASS`/`FAIL`) plus a hazard grade from the category severity. | |
| ### Reading the leaderboard | |
| Each plugin has a deliberately different coverage profile, which is the point of the | |
| framework: | |
| - `sanitizer` kills injected instructions and unverifiable claims β but is blind to | |
| ranking manipulation and metadata exposure. | |
| - `rag_guard` removes poisoned listings at retrieval β powerful, but it over-blocks, | |
| and it cannot see a fabricated spec that looks exactly like a real one. | |
| - `policy_guard` only rewrites the answer β it stops PII leaks and phishing redirects | |
| and nothing else. | |
| **Product hallucination is the category no rule-based defense solves.** Fabricated but | |
| plausible specs ("30-day battery life", "#1 best seller", "50,000 units sold") are | |
| lexically indistinguishable from real ones; closing that gap needs grounding and | |
| external verification, which is the open research direction. | |
| ### Limitations | |
| The agent under test is a **deterministic mock**, not a real LLM. Absolute ASR values | |
| therefore describe this mock; what transfers to real agents is the *relative* coverage | |
| profile of the defenses and the taxonomy itself. The full framework adds an optional | |
| LLM judge channel and LangGraph / OpenAI-Agents adapters. | |
| ### Links | |
| - Dataset β [NoahIsARider/RecRedTeam-ASI-Bench](https://huggingface.co/datasets/NoahIsARider/RecRedTeam-ASI-Bench) | |
| - Framework β [github.com/NoahIsARider/RecRedTeam](https://github.com/NoahIsARider/RecRedTeam) | |
| - Author β [noahisarider.github.io](https://noahisarider.github.io/) | |
| <sub>Defensive research only. Attack cases run against a local simulated catalog, never | |
| against real systems.</sub> | |
| """ | |
| ) | |
| demo.launch() | |