"""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"
  • {_esc(r)}
    • " for r in result["reasons"]) or ( "
  • no attack evidence found in the answer
    • " ) return ( f"
    " f"
    " f"{result['verdict']}  ·  hazard: {result['hazard']}
    " f"" f"
    " ) def _table(headers: List[str], rows: List[List[str]]) -> str: head = "".join(f"{_esc(h)}" for h in headers) body = "".join( "" + "".join( f"{_esc(c)}" if i == 0 else f"{_esc(c)}" for i, c in enumerate(row) ) + "" for row in rows ) return ( "
    " f"{head}{body}
    " ) _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 = ["
    "] for label, value in rows: pct = max(0.0, min(1.0, value)) * 100 hue = int(120 - (pct / 100) * 120) out.append( "
    " f"
    {_esc(label)}
    " "
    " f"
    " f"
    " f"{pct:.1f}%
    " ) out.append("
    ") 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"{case.severity} · " 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']} | {sev} | " 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).", "", "
    Full descriptions", "", ] for t in TAXONOMY: lines.append(f"- **{t['name']}** — {t['description']}") lines += ["", "
    "] 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 "

    Select at least one category.

    ", "", "" 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/) Defensive research only. Attack cases run against a local simulated catalog, never against real systems. """ ) demo.launch()