app.py

"""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']} &nbsp;·&nbsp; 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()