diff --git a/hackagent/server/dashboard/_reports_mixin.py b/hackagent/server/dashboard/_reports_mixin.py index 03274595..91104a38 100644 --- a/hackagent/server/dashboard/_reports_mixin.py +++ b/hackagent/server/dashboard/_reports_mixin.py @@ -182,6 +182,8 @@ def _render_history_goal_detail( self._render_fc_goal_card(row, data, detail_mode=True) # type: ignore[arg-type] else: self._render_tfc_goal_card(row, data, detail_mode=True) # type: ignore[arg-type] + elif ha in ("indirect_prompt_injection", "rag"): + self._render_indirect_injection_view(row, data) # type: ignore[arg-type] else: _req, _resp, _gr_evt = data # type: ignore[misc] self._render_generic_goal_card( diff --git a/hackagent/server/dashboard/_run_history_results_mixin.py b/hackagent/server/dashboard/_run_history_results_mixin.py index 631dd585..a1e0eb90 100644 --- a/hackagent/server/dashboard/_run_history_results_mixin.py +++ b/hackagent/server/dashboard/_run_history_results_mixin.py @@ -44,6 +44,68 @@ class DashboardRunHistoryResultsMixin: """History run results view (_open_run_history_results).""" + def _build_history_goal_detail_data( + self, + attack_type_str: str, + rows: list[dict], + static_template_traces_map: dict[str, list[dict]], + bon_traces_map: dict[str, list[dict]], + generic_traces_map: dict[str, list[dict]], + ) -> dict[str, object]: + """Pre-parse per-row detail data keyed by result id for History goals. + + Dispatches on the run's attack type to the matching ``_parse_*_traces`` + helper. RAG (and its legacy ``indirect_prompt_injection`` alias) skip + parsing and keep the raw serialized traces, since + ``_render_indirect_injection_view`` consumes them directly. Anything + else falls back to a generic request/response extraction. + """ + atk = attack_type_str.lower() + detail_data: dict[str, object] = {} + for row in rows: + rid = str(row.get("id") or "") + if atk in ("static_template", "statictemplate"): + t = static_template_traces_map.get(rid, []) + detail_data[rid] = self._parse_static_template_traces( + t, str(row.get("goal") or "") + ) + elif atk == "bon": + t = bon_traces_map.get(rid, []) + detail_data[rid] = self._parse_bon_traces(t) + elif atk == "pap": + t = generic_traces_map.get(rid, []) + detail_data[rid] = self._parse_pap_traces(t) + elif atk == "pair": + t = generic_traces_map.get(rid, []) + detail_data[rid] = self._parse_pair_traces(t) + elif atk == "crescendo": + t = generic_traces_map.get(rid, []) + detail_data[rid] = self._parse_crescendo_traces(t) + elif atk == "tap": + t = generic_traces_map.get(rid, []) + detail_data[rid] = self._parse_tap_traces(t) + elif atk == "advprefix": + t = generic_traces_map.get(rid, []) + detail_data[rid] = self._parse_advprefix_traces(t) + elif atk == "autodanturbo": + t = generic_traces_map.get(rid, []) + detail_data[rid] = self._parse_autodan_traces(t) + elif atk == "mml": + t = generic_traces_map.get(rid, []) + detail_data[rid] = self._parse_mml_traces(t) + elif atk in ("fc", "tfc"): + t = generic_traces_map.get(rid, []) + if atk == "fc": + detail_data[rid] = self._parse_fc_traces(t) + else: + detail_data[rid] = self._parse_tfc_traces(t) + elif atk in ("indirect_prompt_injection", "rag"): + detail_data[rid] = generic_traces_map.get(rid, []) + else: + t = generic_traces_map.get(rid, []) + detail_data[rid] = self._extract_prompt_response_from_traces(t) + return detail_data + async def _open_run_history_results(self, run: dict) -> None: """Open the compact results list in a non-modal side dialog.""" run_id_raw = str(run.get("id") or "") @@ -1431,50 +1493,13 @@ async def _dl_hcr(): if all_items and self.history_results_list_area is not None: # ── Pre-parse detail data for all rows ───────────── - _h_atk = attack_type_str.lower() - _h_detail_data: dict[str, object] = {} - for _row in new_rows: - _rid = str(_row.get("id") or "") - if _h_atk in ("static_template", "statictemplate"): - _t = static_template_traces_map_hr.get(_rid, []) - _h_detail_data[_rid] = self._parse_static_template_traces( - _t, str(_row.get("goal") or "") - ) - elif _h_atk == "bon": - _t = bon_traces_map_hr.get(_rid, []) - _h_detail_data[_rid] = self._parse_bon_traces(_t) - elif _h_atk == "pap": - _t = generic_traces_map_hr.get(_rid, []) - _h_detail_data[_rid] = self._parse_pap_traces(_t) - elif _h_atk == "pair": - _t = generic_traces_map_hr.get(_rid, []) - _h_detail_data[_rid] = self._parse_pair_traces(_t) - elif _h_atk == "crescendo": - _t = generic_traces_map_hr.get(_rid, []) - _h_detail_data[_rid] = self._parse_crescendo_traces(_t) - elif _h_atk == "tap": - _t = generic_traces_map_hr.get(_rid, []) - _h_detail_data[_rid] = self._parse_tap_traces(_t) - elif _h_atk == "advprefix": - _t = generic_traces_map_hr.get(_rid, []) - _h_detail_data[_rid] = self._parse_advprefix_traces(_t) - elif _h_atk == "autodanturbo": - _t = generic_traces_map_hr.get(_rid, []) - _h_detail_data[_rid] = self._parse_autodan_traces(_t) - elif _h_atk == "mml": - _t = generic_traces_map_hr.get(_rid, []) - _h_detail_data[_rid] = self._parse_mml_traces(_t) - elif _h_atk in ("fc", "tfc"): - _t = generic_traces_map_hr.get(_rid, []) - if _h_atk == "fc": - _h_detail_data[_rid] = self._parse_fc_traces(_t) - else: - _h_detail_data[_rid] = self._parse_tfc_traces(_t) - else: - _t = generic_traces_map_hr.get(_rid, []) - _h_detail_data[_rid] = ( - self._extract_prompt_response_from_traces(_t) - ) # returns (req, resp, guardrail_event) + _h_detail_data = self._build_history_goal_detail_data( + attack_type_str, + new_rows, + static_template_traces_map_hr, + bon_traces_map_hr, + generic_traces_map_hr, + ) # Store for filter re-rendering self._history_goal_rows = new_rows diff --git a/hackagent/server/dashboard/_trace_analysis_mixin.py b/hackagent/server/dashboard/_trace_analysis_mixin.py index daab8ffa..e189eb7e 100644 --- a/hackagent/server/dashboard/_trace_analysis_mixin.py +++ b/hackagent/server/dashboard/_trace_analysis_mixin.py @@ -557,6 +557,8 @@ async def _load_attack_specific_traces( else: detail_data = self._parse_tfc_traces(serialized_traces) self._render_tfc_goal_card(row, detail_data, detail_mode=True) + elif atk in ("indirect_prompt_injection", "rag"): + self._render_indirect_injection_view(row, serialized_traces) else: req_text, resp_text, _generic_guardrail = ( self._extract_prompt_response_from_traces(serialized_traces) diff --git a/tests/unit/cli/tui/test_view_snapshots.py b/tests/unit/cli/tui/test_view_snapshots.py index 2c67c63d..84f7db4d 100644 --- a/tests/unit/cli/tui/test_view_snapshots.py +++ b/tests/unit/cli/tui/test_view_snapshots.py @@ -8,9 +8,17 @@ layout. Regenerate the snapshots with:: uv run pytest tests/unit/cli/tui --snapshot-update + +``snap_compare`` also accepts a file path, but ``pytest-textual-snapshot`` +resolves it to an absolute path and hands it to Textual's ``import_app``, +which runs it through ``shlex.split``. When the repository checkout lives +under a directory containing a space (e.g. ``.../VS Code/hackagent``), that +split mangles the path and the import fails with ``No module named ...``. +Loading the app class ourselves and passing a fresh instance sidesteps +``import_app``/``shlex`` entirely. """ -import shlex +import importlib.util from pathlib import Path import pytest @@ -24,38 +32,29 @@ _NARROW_TERMINAL = (80, 24) -@pytest.fixture(autouse=True) -def _quote_app_path_for_import(monkeypatch): - """Work around a ``textual`` bug when the repo path contains spaces. +def _load_app_instance(app_file: str, class_name: str): + """Import ``app_file`` fresh and return a new instance of ``class_name``. - ``pytest_textual_snapshot.snap_compare`` resolves the app to an absolute - path and hands the raw string to ``textual._import_app.import_app``, - which immediately runs it through ``shlex.split``. Any space in the path - (e.g. a checkout under a directory like ``.../VS Code/...``) is then - parsed as an argument separator, so the importer looks for a module - named after just the first path fragment. Quoting the path before it - reaches ``shlex.split`` keeps it intact as a single token. + A fresh module/instance per call avoids reusing an already-run Textual + ``App`` object across parametrized cases that share the same file. """ - from textual import _import_app as textual_import_app - - original_import_app = textual_import_app.import_app - - def _patched_import_app(import_name: str): - if import_name.endswith(".py") and " " in import_name: - import_name = shlex.quote(import_name) - return original_import_app(import_name) - - monkeypatch.setattr(textual_import_app, "import_app", _patched_import_app) + path = _APPS / app_file + spec = importlib.util.spec_from_file_location(path.stem, path) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return getattr(module, class_name)() @pytest.mark.parametrize( - ("app_file", "terminal_size"), + ("app_file", "class_name", "terminal_size"), [ - ("attacks_tab_app.py", _LARGE_TERMINAL), - ("results_tab_app.py", _LARGE_TERMINAL), - ("attacks_tab_app.py", _NARROW_TERMINAL), + ("attacks_tab_app.py", "AttacksTabApp", _LARGE_TERMINAL), + ("results_tab_app.py", "ResultsTabApp", _LARGE_TERMINAL), + ("attacks_tab_app.py", "AttacksTabApp", _NARROW_TERMINAL), ], ids=["attacks-large", "results-large", "attacks-narrow"], ) -def test_view_renders(snap_compare, app_file, terminal_size): - assert snap_compare(_APPS / app_file, terminal_size=terminal_size) +def test_view_renders(snap_compare, app_file, class_name, terminal_size): + app_instance = _load_app_instance(app_file, class_name) + assert snap_compare(app_instance, terminal_size=terminal_size) diff --git a/tests/unit/server/dashboard/test_rag_history_dispatch.py b/tests/unit/server/dashboard/test_rag_history_dispatch.py new file mode 100644 index 00000000..f438ae3b --- /dev/null +++ b/tests/unit/server/dashboard/test_rag_history_dispatch.py @@ -0,0 +1,278 @@ +# Copyright 2026 - AI4I. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for the RAG/indirect-injection dispatch branches in the History and +Reports views. + +These lock in the fix for a regression where History/Reports goal detail +dispatch dropped the ``rag`` (and legacy ``indirect_prompt_injection``) attack +type branch, silently falling back to the generic goal card and losing the +poisoning/query panel visualization. +""" + +import asyncio +import unittest +from unittest.mock import MagicMock, patch +from uuid import uuid4 + +from hackagent.server.dashboard import _trace_analysis_mixin +from hackagent.server.dashboard._reports_mixin import DashboardReportsMixin +from hackagent.server.dashboard._run_history_results_mixin import ( + DashboardRunHistoryResultsMixin, +) +from hackagent.server.dashboard._trace_analysis_mixin import ( + DashboardTraceAnalysisMixin, +) + + +class _ReportsPage(DashboardReportsMixin): + """Minimal double exposing only what ``_render_history_goal_detail`` needs.""" + + def __init__(self) -> None: + self.rendered_indirect_injection_calls: list[tuple] = [] + + def _render_indirect_injection_view(self, row, traces) -> None: + self.rendered_indirect_injection_calls.append((row, traces)) + + +class TestRenderHistoryGoalDetailDispatchesRag(unittest.TestCase): + def test_rag_attack_type_uses_indirect_injection_view(self): + page = _ReportsPage() + row = {"id": "r1", "goal": "leak secrets"} + traces = [{"content": {"step_name": "Document Poisoning"}}] + + page._render_history_goal_detail(row, traces, "rag") + + self.assertEqual(page.rendered_indirect_injection_calls, [(row, traces)]) + + def test_legacy_indirect_prompt_injection_alias_uses_same_view(self): + page = _ReportsPage() + row = {"id": "r2", "goal": "leak secrets"} + traces = [{"content": {"step_name": "RAG Query #1"}}] + + page._render_history_goal_detail(row, traces, "indirect_prompt_injection") + + self.assertEqual(page.rendered_indirect_injection_calls, [(row, traces)]) + + def test_is_case_insensitive(self): + page = _ReportsPage() + row = {"id": "r3"} + traces = [] + + page._render_history_goal_detail(row, traces, "RAG") + + self.assertEqual(len(page.rendered_indirect_injection_calls), 1) + + +class _HistoryResultsPage(DashboardRunHistoryResultsMixin): + """Minimal double for ``_build_history_goal_detail_data``.""" + + +class TestBuildHistoryGoalDetailDataForRag(unittest.TestCase): + def test_rag_keeps_raw_serialized_traces(self): + page = _HistoryResultsPage() + rows = [{"id": "res-1", "goal": "g1"}] + generic_map = {"res-1": [{"content": {"step_name": "RAG Query #1"}}]} + + result = page._build_history_goal_detail_data("rag", rows, {}, {}, generic_map) + + self.assertEqual(result, {"res-1": generic_map["res-1"]}) + + def test_legacy_alias_also_keeps_raw_traces(self): + page = _HistoryResultsPage() + rows = [{"id": "res-2"}] + generic_map = {"res-2": [{"content": {"step_name": "Document Poisoning"}}]} + + result = page._build_history_goal_detail_data( + "indirect_prompt_injection", rows, {}, {}, generic_map + ) + + self.assertEqual(result, {"res-2": generic_map["res-2"]}) + + def test_missing_traces_default_to_empty_list(self): + page = _HistoryResultsPage() + rows = [{"id": "missing"}] + + result = page._build_history_goal_detail_data("rag", rows, {}, {}, {}) + + self.assertEqual(result, {"missing": []}) + + +class _AllParsersHistoryResultsPage(DashboardRunHistoryResultsMixin): + """Double with a stub for every ``_parse_*_traces`` collaborator. + + Used to exercise *every* branch of ``_build_history_goal_detail_data`` + (not just the rag/indirect_prompt_injection one), since the method + dispatches on attack type to sibling mixins normally composed onto the + real dashboard page class. + """ + + def __init__(self) -> None: + self.calls: list[tuple] = [] + + def _parse_static_template_traces(self, traces, goal=""): + self.calls.append(("static_template", traces, goal)) + return "ST_RESULT" + + def _parse_bon_traces(self, traces): + self.calls.append(("bon", traces)) + return "BON_RESULT" + + def _parse_pap_traces(self, traces): + self.calls.append(("pap", traces)) + return "PAP_RESULT" + + def _parse_pair_traces(self, traces): + self.calls.append(("pair", traces)) + return "PAIR_RESULT" + + def _parse_tap_traces(self, traces): + self.calls.append(("tap", traces)) + return "TAP_RESULT" + + def _parse_advprefix_traces(self, traces): + self.calls.append(("advprefix", traces)) + return "ADVPREFIX_RESULT" + + def _parse_autodan_traces(self, traces): + self.calls.append(("autodan", traces)) + return "AUTODAN_RESULT" + + def _parse_mml_traces(self, traces): + self.calls.append(("mml", traces)) + return "MML_RESULT" + + def _parse_fc_traces(self, traces): + self.calls.append(("fc", traces)) + return "FC_RESULT" + + def _parse_tfc_traces(self, traces): + self.calls.append(("tfc", traces)) + return "TFC_RESULT" + + def _extract_prompt_response_from_traces(self, traces): + self.calls.append(("generic", traces)) + return ("req", "resp", None) + + +class TestBuildHistoryGoalDetailDataAllBranches(unittest.TestCase): + """Covers every dispatch branch of ``_build_history_goal_detail_data``, + not only the rag one, so the whole extracted method is exercised.""" + + def setUp(self): + self.rows = [{"id": "r1", "goal": "g1"}] + self.static_template_map = {"r1": ["st-trace"]} + self.bon_map = {"r1": ["bon-trace"]} + self.generic_map = {"r1": ["generic-trace"]} + + def _build(self, attack_type_str): + page = _AllParsersHistoryResultsPage() + result = page._build_history_goal_detail_data( + attack_type_str, + self.rows, + self.static_template_map, + self.bon_map, + self.generic_map, + ) + return page, result + + def test_static_template(self): + page, result = self._build("static_template") + self.assertEqual(result, {"r1": "ST_RESULT"}) + self.assertEqual(page.calls, [("static_template", ["st-trace"], "g1")]) + + def test_statictemplate_alias(self): + page, result = self._build("statictemplate") + self.assertEqual(result, {"r1": "ST_RESULT"}) + + def test_bon(self): + page, result = self._build("bon") + self.assertEqual(result, {"r1": "BON_RESULT"}) + self.assertEqual(page.calls, [("bon", ["bon-trace"])]) + + def test_pap(self): + page, result = self._build("pap") + self.assertEqual(result, {"r1": "PAP_RESULT"}) + self.assertEqual(page.calls, [("pap", ["generic-trace"])]) + + def test_pair(self): + page, result = self._build("pair") + self.assertEqual(result, {"r1": "PAIR_RESULT"}) + + def test_tap(self): + page, result = self._build("tap") + self.assertEqual(result, {"r1": "TAP_RESULT"}) + + def test_advprefix(self): + page, result = self._build("advprefix") + self.assertEqual(result, {"r1": "ADVPREFIX_RESULT"}) + + def test_autodanturbo(self): + page, result = self._build("autodanturbo") + self.assertEqual(result, {"r1": "AUTODAN_RESULT"}) + + def test_mml(self): + page, result = self._build("mml") + self.assertEqual(result, {"r1": "MML_RESULT"}) + + def test_fc(self): + page, result = self._build("fc") + self.assertEqual(result, {"r1": "FC_RESULT"}) + + def test_tfc(self): + page, result = self._build("tfc") + self.assertEqual(result, {"r1": "TFC_RESULT"}) + + def test_unknown_attack_type_falls_back_to_generic(self): + page, result = self._build("some_unknown_attack") + self.assertEqual(result, {"r1": ("req", "resp", None)}) + self.assertEqual(page.calls, [("generic", ["generic-trace"])]) + + +class _TraceAnalysisPage(DashboardTraceAnalysisMixin): + """Minimal double for ``_load_attack_specific_traces``.""" + + def __init__(self, traces_by_result: dict) -> None: + self._traces_by_result = traces_by_result + self.backend = MagicMock() + self.backend.list_traces.side_effect = lambda result_id: ( + self._traces_by_result.get(str(result_id), []) + ) + self.rendered_indirect_injection_calls: list[tuple] = [] + + def _render_indirect_injection_view(self, row, traces) -> None: + self.rendered_indirect_injection_calls.append((row, traces)) + + +class TestLoadAttackSpecificTracesDispatchesRag(unittest.TestCase): + def test_rag_attack_renders_indirect_injection_view(self): + result_id = uuid4() + raw_trace = {"content": {"step_name": "RAG Query #1"}} + page = _TraceAnalysisPage({str(result_id): [raw_trace]}) + row = {"id": str(result_id), "goal": "leak secrets"} + container = MagicMock() + + with patch.object(_trace_analysis_mixin, "_serialize", lambda t: t): + asyncio.run(page._load_attack_specific_traces(row, container, "rag")) + + self.assertEqual(page.rendered_indirect_injection_calls, [(row, [raw_trace])]) + + def test_legacy_alias_renders_indirect_injection_view(self): + result_id = uuid4() + raw_trace = {"content": {"step_name": "Document Poisoning"}} + page = _TraceAnalysisPage({str(result_id): [raw_trace]}) + row = {"id": str(result_id)} + container = MagicMock() + + with patch.object(_trace_analysis_mixin, "_serialize", lambda t: t): + asyncio.run( + page._load_attack_specific_traces( + row, container, "indirect_prompt_injection" + ) + ) + + self.assertEqual(page.rendered_indirect_injection_calls, [(row, [raw_trace])]) + + +if __name__ == "__main__": + unittest.main()