From d00f673669a37da5ae7626a688f8c19131758e77 Mon Sep 17 00:00:00 2001 From: Richard Lundeen Date: Tue, 14 Jul 2026 11:25:43 -0700 Subject: [PATCH 01/10] FIX: Aligning scenarios with technique registry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Drop the redundant `core`-tag gates from RapidResponse and AdversarialBenchmark so scenarios expose whatever techniques the active initializer has registered, rather than re-narrowing to `core`. - rapid_response: remove `available=TagQuery.all('core')` — the pool is now every registered factory (matches the leakage pattern and the published blog, which lists the extra-only `pair` technique). - benchmark: drop the `'core' in technique_tags` clause but keep the genuine capability filters (`uses_adversarial` and `adversarial_chat is None`), so extra adversarial techniques such as `pair`/`violent_durian` can participate in the model sweep. - Update the benchmark test snapshot to mirror the production predicate. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: e0bc715e-cd3a-480c-afe8-5cac8339224f --- pyrit/scenario/scenarios/airt/rapid_response.py | 6 +++--- pyrit/scenario/scenarios/benchmark/adversarial.py | 10 ++++++---- tests/unit/scenario/benchmark/test_adversarial.py | 6 +++--- 3 files changed, 12 insertions(+), 10 deletions(-) diff --git a/pyrit/scenario/scenarios/airt/rapid_response.py b/pyrit/scenario/scenarios/airt/rapid_response.py index 8362d53595..cde18c4b9b 100644 --- a/pyrit/scenario/scenarios/airt/rapid_response.py +++ b/pyrit/scenario/scenarios/airt/rapid_response.py @@ -35,8 +35,9 @@ def _build_rapid_response_technique() -> type[ScenarioTechnique]: """ Build the RapidResponse technique class dynamically from the registered factories. - Reads the singleton ``AttackTechniqueRegistry`` and filters to factories - tagged ``core``. + Reads every technique registered in the singleton ``AttackTechniqueRegistry`` + and exposes all of them. Which techniques are available is decided by the + active initializer (the registration gate), not narrowed again here. Returns: type[ScenarioTechnique]: The dynamically generated technique enum class. @@ -50,7 +51,6 @@ def _build_rapid_response_technique() -> type[ScenarioTechnique]: return AttackTechniqueRegistry.build_technique_class_from_factories( # type: ignore[ty:invalid-return-type] class_name="RapidResponseTechnique", factories=factories, - available=TagQuery.all("core"), aggregate_tags={ "single_turn": TagQuery.any_of("single_turn"), "multi_turn": TagQuery.any_of("multi_turn"), diff --git a/pyrit/scenario/scenarios/benchmark/adversarial.py b/pyrit/scenario/scenarios/benchmark/adversarial.py index a62bda6a3b..02128c321f 100644 --- a/pyrit/scenario/scenarios/benchmark/adversarial.py +++ b/pyrit/scenario/scenarios/benchmark/adversarial.py @@ -35,12 +35,14 @@ def _build_benchmark_technique() -> type[ScenarioTechnique]: """ Build the ``BenchmarkTechnique`` enum from the registered factory catalog. - Reads ``core`` adversarial-capable factories from the + Reads adversarial-capable factories from the ``AttackTechniqueRegistry`` singleton and passes them to ``build_technique_class_from_factories``. Factories that bake their own ``adversarial_chat`` are excluded — the benchmark sweeps each technique across the user-supplied targets, which is incompatible with a technique - that pins its own adversarial target. The resulting enum has one + that pins its own adversarial target. Which techniques are registered is + decided by the active initializer (the registration gate); this scenario + does not narrow the pool further by group. The resulting enum has one concrete member per factory (e.g. ``red_teaming``, ``tap``, ``crescendo_simulated``) plus ``default`` / ``light`` / ``single_turn`` / ``multi_turn`` aggregates derived from each factory's ``technique_tags``. @@ -56,7 +58,7 @@ def _build_benchmark_technique() -> type[ScenarioTechnique]: factories = [ factory for factory in registry.get_factories_or_raise().values() - if factory.uses_adversarial and "core" in factory.technique_tags and factory.adversarial_chat is None + if factory.uses_adversarial and factory.adversarial_chat is None ] return AttackTechniqueRegistry.build_technique_class_from_factories( # type: ignore[ty:invalid-return-type] class_name="BenchmarkTechnique", @@ -82,7 +84,7 @@ class AdversarialBenchmark(Scenario): At run time, ``_build_atomic_attacks_async`` performs the ``(technique × adversarial_target × dataset)`` cross-product: for each - selected adversarial-capable ``core`` factory in the + selected adversarial-capable factory in the ``AttackTechniqueRegistry`` and each requested target, it calls ``factory.create(adversarial_chat=...)`` with the resolved target — no global registry mutation. The resulting diff --git a/tests/unit/scenario/benchmark/test_adversarial.py b/tests/unit/scenario/benchmark/test_adversarial.py index 8b1cbf2ecc..dd5996c5f2 100644 --- a/tests/unit/scenario/benchmark/test_adversarial.py +++ b/tests/unit/scenario/benchmark/test_adversarial.py @@ -15,7 +15,7 @@ These tests cover the new contract: * Class metadata (VERSION, BASELINE policy, defaults). * Technique enum is built from registered factories with ``uses_adversarial=True`` - and the ``core`` technique tag; ``light`` aggregate preserves the + that do not bake their own ``adversarial_chat``; ``light`` aggregate preserves the source ``light`` tag (excludes ``tap`` / ``crescendo_simulated``). * ``supported_parameters`` declares ``adversarial_targets: list[str]``. * ``_resolve_adversarial_targets`` raises with available names on typos. @@ -76,7 +76,7 @@ def _build_benchmarkable_factories_snapshot() -> list: factories = build_technique_factories() finally: TargetRegistry.reset_registry_singleton() - return [f for f in factories if f.uses_adversarial and "core" in f.technique_tags] + return [f for f in factories if f.uses_adversarial and f.adversarial_chat is None] _BENCHMARKABLE_FACTORIES = _build_benchmarkable_factories_snapshot() @@ -202,7 +202,7 @@ class TestAdversarialBenchmarkTechnique: """Tests for ``_build_benchmark_technique`` using the registry-based factory API.""" def test_technique_built_from_registered_adversarial_factories(self): - """Each registered ``core`` adversarial factory produces one concrete enum member.""" + """Each registered adversarial factory produces one concrete enum member.""" technique_cls = _build_benchmark_technique() aggregate_names = {"all"} | technique_cls.get_aggregate_tags() concrete_members = [m for m in technique_cls if m.value not in aggregate_names] From 81cfcd1e9b55d152af00f36589c85df55ea2dabe Mon Sep 17 00:00:00 2001 From: Richard Lundeen Date: Tue, 14 Jul 2026 12:37:21 -0700 Subject: [PATCH 02/10] FEAT: Make catalog tags synonymous with selectable aggregates Auto-promote every catalog tag present in a scenario's technique pool into a selectable aggregate, so `--techniques ` (e.g. `core`, or a custom `airt_internal`) expands to every technique carrying that tag. `all` and `default` stay reserved synthetic aggregates; a tag that collides with a technique name stays a concrete technique (name selection wins). - attack_technique_registry: derive aggregates from pool tags; make aggregate_tags optional (escape hatch for compound/renamed aggregates only). - Strip the now-redundant 1:1 aggregate_tags from all six call sites (rapid_response, cyber, text_adaptive, leakage, benchmark, doctor) and drop unused TagQuery imports. - cyber: drop the curation gate so ALL exposes the full registered pool while DEFAULT stays curated to red_teaming (aligns with rapid_response/benchmark). - Add a shared invariant that every catalog tag is a selectable aggregate. - Fix benchmark test mock factory (adversarial_chat=None) so it lands in the benchmarkable pool now that `light` is auto-derived rather than declared. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: e0bc715e-cd3a-480c-afe8-5cac8339224f --- .../components/attack_technique_registry.py | 44 ++++++++++++++----- .../scenarios/adaptive/text_adaptive.py | 5 --- pyrit/scenario/scenarios/airt/cyber.py | 33 +++++--------- pyrit/scenario/scenarios/airt/leakage.py | 5 --- .../scenario/scenarios/airt/rapid_response.py | 5 --- .../scenarios/benchmark/adversarial.py | 11 ++--- pyrit/scenario/scenarios/garak/doctor.py | 1 - tests/unit/scenario/airt/test_cyber.py | 28 +++++++----- .../scenario/benchmark/test_adversarial.py | 1 + .../test_scenario_technique_invariants.py | 21 +++++++++ 10 files changed, 87 insertions(+), 67 deletions(-) diff --git a/pyrit/registry/components/attack_technique_registry.py b/pyrit/registry/components/attack_technique_registry.py index ffa0bbbca0..275f4bf1b1 100644 --- a/pyrit/registry/components/attack_technique_registry.py +++ b/pyrit/registry/components/attack_technique_registry.py @@ -177,7 +177,7 @@ def build_technique_class_from_factories( *, class_name: str, factories: list[AttackTechniqueFactory], - aggregate_tags: dict[str, TagQuery], + aggregate_tags: dict[str, TagQuery] | None = None, available: TagQuery | None = None, default: TagQuery | None = None, default_technique_names: set[str] | None = None, @@ -189,6 +189,9 @@ def build_technique_class_from_factories( - An ``ALL`` aggregate member (always included). - A ``DEFAULT`` aggregate member when a default selection is provided. - Additional aggregate members from ``aggregate_tags`` keys. + - An aggregate member for every other catalog tag present in the pool, so + tags and aggregates are synonymous: selecting a tag (e.g. ``core`` or a + custom ``airt_internal``) expands to every technique carrying it. - One technique member per *available* factory, with tags from the factory. The three selection roles are all expressed the same way — as tag queries @@ -197,25 +200,29 @@ def build_technique_class_from_factories( - **available** (the pool): ``available`` filters ``factories`` to the techniques this scenario exposes. When ``None`` the whole ``factories`` list is the pool (back-compatible). - - **aggregates**: named ``TagQuery`` presets (e.g. ``single_turn``); each is - evaluated only over the pool, so every aggregate is a subset of available. + - **aggregates**: selectable tag groups. Every catalog tag present in the + pool becomes one automatically; ``aggregate_tags`` adds explicit ones for + queries a single tag can't express. Each is evaluated only over the pool, + so every aggregate is a subset of available. - **default**: what runs when the caller selects nothing. Given as a ``TagQuery`` (``default``) and/or explicit ``default_technique_names``; both are intersected with the pool, so ``DEFAULT`` is always a subset of available. - ``default`` is deliberately **not** an intrinsic technique tag: what runs by - default differs per scenario. A scenario selects its default set via a query - or by name so the same technique can be default for one scenario and not - another, without a catalog-wide tag. + ``default`` is chosen per-scenario: a scenario picks its default set by query + or by name, so the same technique can be the default for one scenario and not + another. Args: class_name (str): Name for the generated enum class. factories (list[AttackTechniqueFactory]): Candidate technique factories. Filtered by ``available`` to form the pool of enum members. - aggregate_tags (dict[str, TagQuery]): Maps aggregate member names to a - ``TagQuery`` that selects which pool techniques belong to the aggregate. - An ``ALL`` aggregate (expanding to all pool techniques) is always added. + aggregate_tags (dict[str, TagQuery] | None): Optional explicit aggregate + definitions, mapping an aggregate member name to a ``TagQuery`` over the + pool. Only needed for aggregates a single tag can't express — a compound + query (AND/OR/NOT) or an aggregate whose name is not itself a tag. Simple + one-to-one tag aggregates are auto-derived (see above), so this can be + omitted. An ``ALL`` aggregate is always added. available (TagQuery | None): Query selecting which of ``factories`` are available for this scenario (the pool). ``None`` means all of them. default (TagQuery | None): Query selecting the pool techniques that form @@ -233,6 +240,8 @@ def build_technique_class_from_factories( """ from pyrit.scenario import ScenarioTechnique + aggregate_tags = aggregate_tags or {} + # available (the pool): filter the candidate factories by the availability query. pool = available.filter(factories) if available is not None else list(factories) @@ -244,7 +253,18 @@ def build_technique_class_from_factories( if default is not None: default_names |= {f.name for f in pool if default.matches(set(f.technique_tags))} - all_aggregate_tag_names = {"all"} | set(aggregate_tags.keys()) + # Auto-promote every catalog tag present in the pool into a selectable + # aggregate, so tags and aggregates are synonymous: selecting a tag expands to + # every technique carrying it. "all" and "default" are reserved synthetic + # aggregates (see above) and are never derived from tags. A tag that collides + # with a technique name stays a concrete technique (name selection wins), and + # tags already declared in aggregate_tags keep their explicit TagQuery. + reserved_aggregate_tags = {"all", "default"} + pool_technique_names = {f.name for f in pool} + pool_tags = {tag for f in pool for tag in f.technique_tags} + auto_aggregate_tags = pool_tags - reserved_aggregate_tags - pool_technique_names - set(aggregate_tags) + + all_aggregate_tag_names = {"all"} | set(aggregate_tags.keys()) | auto_aggregate_tags if default_names: all_aggregate_tag_names.add("default") @@ -256,6 +276,8 @@ def build_technique_class_from_factories( members["DEFAULT"] = ("default", {"default"}) for agg_name in aggregate_tags: members[agg_name.upper()] = (agg_name, {agg_name}) + for agg_name in sorted(auto_aggregate_tags): + members[agg_name.upper()] = (agg_name, {agg_name}) # Technique members from the pool — assign aggregate tags based on TagQuery matching for factory in pool: diff --git a/pyrit/scenario/scenarios/adaptive/text_adaptive.py b/pyrit/scenario/scenarios/adaptive/text_adaptive.py index 30d616516b..537fb7f096 100644 --- a/pyrit/scenario/scenarios/adaptive/text_adaptive.py +++ b/pyrit/scenario/scenarios/adaptive/text_adaptive.py @@ -19,7 +19,6 @@ from pyrit.common import apply_defaults from pyrit.models.parameter import Parameter from pyrit.registry.components.attack_technique_registry import AttackTechniqueRegistry -from pyrit.registry.tag_query import TagQuery from pyrit.scenario.core.dataset_configuration import CompoundDatasetAttackConfiguration, DatasetAttackConfiguration from pyrit.scenario.scenarios.adaptive.adaptive_scenario import AdaptiveScenario @@ -71,10 +70,6 @@ def _build_text_adaptive_technique() -> type[ScenarioTechnique]: return AttackTechniqueRegistry.build_technique_class_from_factories( # type: ignore[return-value, ty:invalid-return-type] class_name="TextAdaptiveTechnique", factories=factories, - aggregate_tags={ - "single_turn": TagQuery.any_of("single_turn"), - "multi_turn": TagQuery.any_of("multi_turn"), - }, default_technique_names={"role_play_movie_script", "many_shot"}, ) diff --git a/pyrit/scenario/scenarios/airt/cyber.py b/pyrit/scenario/scenarios/airt/cyber.py index 51debd1df9..dedd83aa7c 100644 --- a/pyrit/scenario/scenarios/airt/cyber.py +++ b/pyrit/scenario/scenarios/airt/cyber.py @@ -23,12 +23,9 @@ logger = logging.getLogger(__name__) -# Techniques Cyber selects from the shared catalog. Cyber declares its own DEFAULT by naming -# these techniques (see _build_cyber_technique), rather than relying on a catalog-wide ``default`` -# tag. Adding a technique here that is not also in ``_CYBER_DEFAULT_TECHNIQUE_NAMES`` would keep -# it in ALL but out of DEFAULT, breaking the current DEFAULT == ALL invariant guarded by -# test_default_matches_all; keep the two sets in sync if this list grows. -_CYBER_TECHNIQUE_NAMES = {"red_teaming"} +# Cyber curates its DEFAULT run to the technique(s) named here (see _build_cyber_technique). +# The pool of *available* techniques is not narrowed — cyber exposes whatever the active +# initializer has registered (like RapidResponse); the initializer is the single gate. _CYBER_DEFAULT_TECHNIQUE_NAMES = {"red_teaming"} @@ -37,33 +34,25 @@ def _build_cyber_technique() -> type[ScenarioTechnique]: """ Build the Cyber technique class dynamically from the registered technique factories. - Selects only the ``red_teaming`` factory from the singleton - ``AttackTechniqueRegistry``. A plain ``PromptSendingAttack`` baseline is - prepended automatically by ``Scenario._build_baseline_atomic_attack`` via - ``BaselineAttackPolicy.Enabled``. + Exposes every technique registered in the singleton ``AttackTechniqueRegistry``; + which techniques are available is decided by the active initializer, not narrowed + here. A plain ``PromptSendingAttack`` baseline is prepended automatically by + ``Scenario._build_baseline_atomic_attack`` via ``BaselineAttackPolicy.Enabled``. - The ``DEFAULT`` aggregate is the curated default run; for Cyber it expands to the - same single ``red_teaming`` technique as ``ALL``. + The ``DEFAULT`` aggregate is the curated default run — for Cyber it expands to + ``red_teaming`` — while ``ALL`` selects the full registered pool. Returns: type[ScenarioTechnique]: The dynamically generated technique enum class. """ from pyrit.registry.components.attack_technique_registry import AttackTechniqueRegistry - from pyrit.registry.tag_query import TagQuery registry = AttackTechniqueRegistry.get_registry_singleton() - factories = registry.get_factories_or_raise() - cyber_factories = [f for name, f in factories.items() if name in _CYBER_TECHNIQUE_NAMES] + factories = list(registry.get_factories_or_raise().values()) return AttackTechniqueRegistry.build_technique_class_from_factories( # type: ignore[ty:invalid-return-type] class_name="CyberTechnique", - factories=cyber_factories, - aggregate_tags={ - "multi_turn": TagQuery.any_of("multi_turn"), - }, - # Cyber curates a single technique (red_teaming) at the scenario level. It declares that - # as its DEFAULT by name rather than tagging red_teaming ``default`` globally, which would - # alter other scenarios. + factories=factories, default_technique_names=_CYBER_DEFAULT_TECHNIQUE_NAMES, ) diff --git a/pyrit/scenario/scenarios/airt/leakage.py b/pyrit/scenario/scenarios/airt/leakage.py index 0166a36564..492e775b84 100644 --- a/pyrit/scenario/scenarios/airt/leakage.py +++ b/pyrit/scenario/scenarios/airt/leakage.py @@ -10,7 +10,6 @@ from pyrit.common import apply_defaults from pyrit.common.path import SCORER_SEED_PROMPT_PATH from pyrit.registry.components.attack_technique_registry import AttackTechniqueRegistry -from pyrit.registry.tag_query import TagQuery from pyrit.scenario.core.dataset_configuration import DatasetAttackConfiguration from pyrit.scenario.core.matrix_atomic_attack_builder import build_matrix_atomic_attacks from pyrit.scenario.core.scenario import Scenario @@ -67,10 +66,6 @@ def _build_leakage_technique() -> type[ScenarioTechnique]: return AttackTechniqueRegistry.build_technique_class_from_factories( # type: ignore[return-value, ty:invalid-return-type] class_name="LeakageTechnique", factories=all_factories, - aggregate_tags={ - "single_turn": TagQuery.any_of("single_turn"), - "multi_turn": TagQuery.any_of("multi_turn"), - }, default_technique_names={"role_play_movie_script", "many_shot", "first_letter", "image"}, ) diff --git a/pyrit/scenario/scenarios/airt/rapid_response.py b/pyrit/scenario/scenarios/airt/rapid_response.py index cde18c4b9b..424c1c622a 100644 --- a/pyrit/scenario/scenarios/airt/rapid_response.py +++ b/pyrit/scenario/scenarios/airt/rapid_response.py @@ -43,7 +43,6 @@ def _build_rapid_response_technique() -> type[ScenarioTechnique]: type[ScenarioTechnique]: The dynamically generated technique enum class. """ from pyrit.registry.components.attack_technique_registry import AttackTechniqueRegistry - from pyrit.registry.tag_query import TagQuery registry = AttackTechniqueRegistry.get_registry_singleton() factories = list(registry.get_factories_or_raise().values()) @@ -51,10 +50,6 @@ def _build_rapid_response_technique() -> type[ScenarioTechnique]: return AttackTechniqueRegistry.build_technique_class_from_factories( # type: ignore[ty:invalid-return-type] class_name="RapidResponseTechnique", factories=factories, - aggregate_tags={ - "single_turn": TagQuery.any_of("single_turn"), - "multi_turn": TagQuery.any_of("multi_turn"), - }, default_technique_names={"role_play_movie_script", "many_shot"}, ) diff --git a/pyrit/scenario/scenarios/benchmark/adversarial.py b/pyrit/scenario/scenarios/benchmark/adversarial.py index 02128c321f..7266030667 100644 --- a/pyrit/scenario/scenarios/benchmark/adversarial.py +++ b/pyrit/scenario/scenarios/benchmark/adversarial.py @@ -14,7 +14,6 @@ from pyrit.models import AttackOutcome, AttackResult, ObjectiveTargetEvaluationIdentifier, ScenarioResult from pyrit.models.parameter import Parameter from pyrit.registry import AttackTechniqueRegistry, TargetRegistry -from pyrit.registry.tag_query import TagQuery from pyrit.scenario.core.dataset_configuration import DatasetAttackConfiguration from pyrit.scenario.core.matrix_atomic_attack_builder import MatrixAtomicAttackBuilder, resolve_technique_factories from pyrit.scenario.core.scenario import BaselineAttackPolicy, Scenario @@ -44,8 +43,9 @@ def _build_benchmark_technique() -> type[ScenarioTechnique]: decided by the active initializer (the registration gate); this scenario does not narrow the pool further by group. The resulting enum has one concrete member per factory (e.g. ``red_teaming``, ``tap``, - ``crescendo_simulated``) plus ``default`` / ``light`` / ``single_turn`` - / ``multi_turn`` aggregates derived from each factory's ``technique_tags``. + ``crescendo_simulated``), a ``light`` / ``single_turn`` / ``multi_turn`` + aggregate for each catalog tag, and a ``default`` aggregate for the + scenario's default run. The (technique × target) cross-product is materialized lazily in ``AdversarialBenchmark._build_atomic_attacks_async`` from the @@ -63,11 +63,6 @@ def _build_benchmark_technique() -> type[ScenarioTechnique]: return AttackTechniqueRegistry.build_technique_class_from_factories( # type: ignore[ty:invalid-return-type] class_name="BenchmarkTechnique", factories=factories, - aggregate_tags={ - "light": TagQuery.any_of("light"), - "single_turn": TagQuery.any_of("single_turn"), - "multi_turn": TagQuery.any_of("multi_turn"), - }, default_technique_names={"role_play_movie_script", "many_shot"}, ) diff --git a/pyrit/scenario/scenarios/garak/doctor.py b/pyrit/scenario/scenarios/garak/doctor.py index 9c45174471..cd3a206a4f 100644 --- a/pyrit/scenario/scenarios/garak/doctor.py +++ b/pyrit/scenario/scenarios/garak/doctor.py @@ -82,7 +82,6 @@ def _build_doctor_technique() -> type[ScenarioTechnique]: return AttackTechniqueRegistry.build_technique_class_from_factories( # type: ignore[return-value, ty:invalid-return-type] class_name="DoctorTechnique", factories=DOCTOR_FACTORIES, - aggregate_tags={}, default_technique_names={"policy_puppetry", "policy_puppetry_leet"}, ) diff --git a/tests/unit/scenario/airt/test_cyber.py b/tests/unit/scenario/airt/test_cyber.py index 70ff711f10..9051ffa68d 100644 --- a/tests/unit/scenario/airt/test_cyber.py +++ b/tests/unit/scenario/airt/test_cyber.py @@ -149,15 +149,17 @@ def test_default_aggregate_expands_to_red_teaming(self): default_members = strat.expand({strat.DEFAULT}) assert default_members == [strat("red_teaming")] - def test_default_matches_all(self): - """DEFAULT must expand to exactly the same techniques as ALL. + def test_all_is_a_superset_of_default(self): + """ALL exposes the full registered pool while DEFAULT stays curated to red_teaming. - Cyber curates a single technique, so its DEFAULT run is a no-op alias of ALL. Asserting - equality (not just subset) guards against a future technique landing in ALL but being - silently excluded from DEFAULT (or vice versa) once the aggregate wiring changes. + Cyber no longer narrows its available techniques — the initializer is the single + gate — so ALL is strictly broader than the curated DEFAULT run. """ strat = _technique_class() - assert set(strat.expand({strat.DEFAULT})) == set(strat.expand({strat.ALL})) + default_members = set(strat.expand({strat.DEFAULT})) + all_members = set(strat.expand({strat.ALL})) + assert default_members == {strat("red_teaming")} + assert default_members < all_members def test_default_dataset_config_has_malware_dataset(self): config = Cyber()._default_dataset_config @@ -296,23 +298,29 @@ async def _init_and_get_attacks( await scenario.initialize_async() return scenario._atomic_attacks - async def test_all_technique_produces_red_teaming(self, mock_objective_target, mock_objective_scorer): + async def test_all_technique_includes_red_teaming_and_others(self, mock_objective_target, mock_objective_scorer): + """ALL now exposes the full registered pool, not just red_teaming.""" attacks = await self._init_and_get_attacks( mock_objective_target=mock_objective_target, mock_objective_scorer=mock_objective_scorer, techniques=[_technique_class().ALL], ) technique_classes = {type(a.attack_technique.attack) for a in attacks} - assert technique_classes == {RedTeamingAttack} + assert RedTeamingAttack in technique_classes + assert len(technique_classes) > 1 - async def test_multi_turn_technique_produces_red_teaming(self, mock_objective_target, mock_objective_scorer): + async def test_multi_turn_technique_includes_red_teaming_and_others( + self, mock_objective_target, mock_objective_scorer + ): + """MULTI_TURN exposes every registered multi-turn technique, including red_teaming.""" attacks = await self._init_and_get_attacks( mock_objective_target=mock_objective_target, mock_objective_scorer=mock_objective_scorer, techniques=[_technique_class().MULTI_TURN], ) technique_classes = {type(a.attack_technique.attack) for a in attacks} - assert technique_classes == {RedTeamingAttack} + assert RedTeamingAttack in technique_classes + assert len(technique_classes) > 1 async def test_default_technique_produces_red_teaming(self, mock_objective_target, mock_objective_scorer): """Default (DEFAULT) should produce RedTeaming. PromptSendingAttack baseline is diff --git a/tests/unit/scenario/benchmark/test_adversarial.py b/tests/unit/scenario/benchmark/test_adversarial.py index dd5996c5f2..9c0745ba3a 100644 --- a/tests/unit/scenario/benchmark/test_adversarial.py +++ b/tests/unit/scenario/benchmark/test_adversarial.py @@ -126,6 +126,7 @@ def _register_mock_factory(*, name: str, tags: list[str] | None = None, seed_tec factory = MagicMock(spec=AttackTechniqueFactory) factory.name = name factory.uses_adversarial = True + factory.adversarial_chat = None factory.technique_tags = tags if tags is not None else ["core", "light"] factory.seed_technique = seed_technique technique_instance = MagicMock(name="AttackTechnique") diff --git a/tests/unit/scenario/core/test_scenario_technique_invariants.py b/tests/unit/scenario/core/test_scenario_technique_invariants.py index 9d0098a39d..dccbffa678 100644 --- a/tests/unit/scenario/core/test_scenario_technique_invariants.py +++ b/tests/unit/scenario/core/test_scenario_technique_invariants.py @@ -183,6 +183,27 @@ def test_aggregates_are_disjoint_from_techniques(get_technique): assert agg_values.isdisjoint(tech_values) +@pytest.mark.parametrize("get_technique", SCENARIO_STRATEGY_BUILDERS) +def test_catalog_tags_are_selectable_aggregates(get_technique): + """Every catalog tag carried by a pool technique is itself a selectable aggregate + that expands to exactly the techniques carrying it (tags are synonymous with + aggregates). ``all`` and ``default`` are reserved synthetic aggregates, and a tag + that collides with a technique name stays a concrete technique, so both are excluded. + """ + strat = get_technique() + reserved = {"all", "default"} + technique_values = {s.value for s in strat.get_all_techniques()} + catalog_tags = {tag for s in strat.get_all_techniques() for tag in s.tags} - reserved - technique_values + assert catalog_tags, "expected at least one catalog tag to promote" + + aggregate_values = {s.value for s in strat.get_aggregate_techniques()} + for tag in catalog_tags: + assert tag in aggregate_values, f"catalog tag {tag!r} is not a selectable aggregate" + expanded = set(strat.expand({strat(tag)})) + expected = {s for s in strat.get_all_techniques() if tag in s.tags} + assert expanded == expected, f"aggregate {tag!r} did not expand to its tagged techniques" + + @pytest.mark.parametrize("get_technique", SCENARIO_STRATEGY_BUILDERS) def test_expanding_a_technique_returns_itself(get_technique): """Expanding a single non-aggregate technique returns just that technique.""" From d56a89508e08d5344717184dfd8cddaf384bcab5 Mon Sep 17 00:00:00 2001 From: Richard Lundeen Date: Wed, 15 Jul 2026 14:02:59 -0700 Subject: [PATCH 03/10] FIX: Bump broadened scenarios to VERSION 3 and make benchmark default robust Bump cyber, rapid_response, and adversarial benchmark to VERSION = 3 since dropping the ``core`` pool gate broadened their selectable ``all`` pool; ``use_cached`` only matches prior runs at the current VERSION, so stale v2 results must not suppress v3 runs. Make AdversarialBenchmark.__init__ fall back to the ``all`` aggregate when no ``light``-tagged factory survives the pool filter, instead of unconditionally resolving ``technique_class("light")`` which raised for custom technique sets. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: e0bc715e-cd3a-480c-afe8-5cac8339224f --- pyrit/scenario/scenarios/airt/cyber.py | 5 ++++- .../scenario/scenarios/airt/rapid_response.py | 5 ++++- .../scenarios/benchmark/adversarial.py | 15 ++++++++++++--- tests/unit/scenario/airt/test_cyber.py | 4 ++-- .../unit/scenario/airt/test_rapid_response.py | 4 ++-- .../scenario/benchmark/test_adversarial.py | 19 ++++++++++++++++--- 6 files changed, 40 insertions(+), 12 deletions(-) diff --git a/pyrit/scenario/scenarios/airt/cyber.py b/pyrit/scenario/scenarios/airt/cyber.py index dedd83aa7c..a9c56508b5 100644 --- a/pyrit/scenario/scenarios/airt/cyber.py +++ b/pyrit/scenario/scenarios/airt/cyber.py @@ -66,7 +66,10 @@ class Cyber(Scenario): techniques. """ - VERSION: int = 2 + #: Bumped from 2 → 3 by dropping the ``core`` pool gate so the selectable + #: technique pool (and the ``all`` aggregate) reflects whatever the initializer + #: registered. ``use_cached`` only matches prior runs at the current ``VERSION``. + VERSION: int = 3 @classmethod def get_override_composite_scorer_questions_path(cls) -> list[Path]: diff --git a/pyrit/scenario/scenarios/airt/rapid_response.py b/pyrit/scenario/scenarios/airt/rapid_response.py index 424c1c622a..517fa6e934 100644 --- a/pyrit/scenario/scenarios/airt/rapid_response.py +++ b/pyrit/scenario/scenarios/airt/rapid_response.py @@ -62,7 +62,10 @@ class RapidResponse(Scenario): techniques. """ - VERSION: int = 2 + #: Bumped from 2 → 3 by dropping the ``core`` pool gate so the selectable + #: technique pool (and the ``all`` aggregate) reflects whatever the initializer + #: registered. ``use_cached`` only matches prior runs at the current ``VERSION``. + VERSION: int = 3 @apply_defaults def __init__( diff --git a/pyrit/scenario/scenarios/benchmark/adversarial.py b/pyrit/scenario/scenarios/benchmark/adversarial.py index 7266030667..4e9304e5cb 100644 --- a/pyrit/scenario/scenarios/benchmark/adversarial.py +++ b/pyrit/scenario/scenarios/benchmark/adversarial.py @@ -92,9 +92,12 @@ class AdversarialBenchmark(Scenario): #: from a constructor parameter to the ``adversarial_targets`` scenario #: parameter and changed ``atomic_attack_name`` from #: ``{technique}__{model}__{dataset}`` to ``{technique}__{target}_{dataset}``. + #: Bumped from 2 → 3 by dropping the ``core`` pool gate so the selectable + #: technique pool (and therefore the ``all`` aggregate) reflects whatever the + #: initializer registered rather than only core-tagged factories. #: ``use_cached`` only matches against prior runs at the current - #: ``VERSION``; v1 results remain queryable but won't suppress v2 runs. - VERSION: int = 2 + #: ``VERSION``; older results remain queryable but won't suppress v3 runs. + VERSION: int = 3 #: AdversarialBenchmark compares attack-success rates across adversarial models; a baseline #: attack would be model-independent and contribute no signal to the comparison. @@ -174,11 +177,17 @@ def __init__( technique_class = _build_benchmark_technique() + # ``light`` is the curated default run, but it only exists when the pool + # (decided by the active initializer) contains a light-tagged factory. Fall + # back to ``all`` so a custom technique set with no light factory still yields + # a constructible benchmark rather than raising on an absent aggregate. + default_aggregate = "light" if "light" in technique_class.get_aggregate_tags() else "all" + super().__init__( version=self.VERSION, objective_scorer=self._objective_scorer, technique_class=technique_class, - default_technique=technique_class("light"), + default_technique=technique_class(default_aggregate), default_dataset_config=DatasetAttackConfiguration( dataset_names=["harmbench"], max_dataset_size=8, diff --git a/tests/unit/scenario/airt/test_cyber.py b/tests/unit/scenario/airt/test_cyber.py index 9051ffa68d..d29d3b5caa 100644 --- a/tests/unit/scenario/airt/test_cyber.py +++ b/tests/unit/scenario/airt/test_cyber.py @@ -126,8 +126,8 @@ def _make_seed_groups(name: str) -> list[AttackSeedGroup]: class TestCyberBasic: """Tests for Cyber initialization and class properties.""" - def test_version_is_2(self): - assert Cyber.VERSION == 2 + def test_version_is_3(self): + assert Cyber.VERSION == 3 def test_get_technique_class(self): strat = _technique_class() diff --git a/tests/unit/scenario/airt/test_rapid_response.py b/tests/unit/scenario/airt/test_rapid_response.py index 4a04ef3bcf..29d8f4653c 100644 --- a/tests/unit/scenario/airt/test_rapid_response.py +++ b/tests/unit/scenario/airt/test_rapid_response.py @@ -155,8 +155,8 @@ def _make_seed_groups(name: str) -> list[AttackSeedGroup]: class TestRapidResponseBasic: """Tests for RapidResponse initialization and class properties.""" - def test_version_is_2(self): - assert RapidResponse.VERSION == 2 + def test_version_is_3(self): + assert RapidResponse.VERSION == 3 def test_get_technique_class(self, mock_objective_scorer): strat = _technique_class() diff --git a/tests/unit/scenario/benchmark/test_adversarial.py b/tests/unit/scenario/benchmark/test_adversarial.py index 9c0745ba3a..6edcc0dda9 100644 --- a/tests/unit/scenario/benchmark/test_adversarial.py +++ b/tests/unit/scenario/benchmark/test_adversarial.py @@ -155,9 +155,9 @@ async def _build_atomic_attacks(bench: AdversarialBenchmark) -> list: class TestAdversarialBenchmarkMetadata: """Tests for class-level metadata that doesn't depend on any runtime state.""" - def test_version_is_2(self): - """VERSION matches the post-collapse ``atomic_attack_name`` format so cached results still match.""" - assert AdversarialBenchmark.VERSION == 2 + def test_version_is_3(self): + """VERSION bumped to 3 when the ``core`` pool gate was dropped so cached v2 results don't suppress v3 runs.""" + assert AdversarialBenchmark.VERSION == 3 def test_baseline_attack_policy_is_forbidden(self): """A baseline contributes no signal to a model-comparison benchmark, so it is forbidden.""" @@ -241,6 +241,8 @@ def test_technique_excludes_factories_with_baked_adversarial_chat(self): technique_cls = _build_benchmark_technique() member_values = {m.value for m in technique_cls} assert "pinned_adversary" not in member_values + + def test_technique_exposes_tag_aggregates(self): """The technique enum exposes ``light``, ``single_turn``, ``multi_turn`` aggregates.""" technique_cls = _build_benchmark_technique() aggregates = technique_cls.get_aggregate_tags() @@ -306,6 +308,17 @@ def test_skip_cached_can_be_set_true(self): ) assert bench._use_cached is True + def test_construct_without_light_factory_falls_back_to_all(self): + """A pool with no ``light``-tagged factory must still construct, defaulting to ``all``.""" + AttackTechniqueRegistry.reset_registry_singleton() + _build_benchmark_technique.cache_clear() + _register_mock_factory(name="narrow_tap", tags=["airt_internal", "multi_turn"]) + + bench = AdversarialBenchmark(objective_scorer=MagicMock(spec=TrueFalseScorer)) + + assert "light" not in bench._technique_class.get_aggregate_tags() + assert bench._default_technique.value == "all" + # --------------------------------------------------------------------------- # _resolve_adversarial_targets From 160248bd813cbafe823b20c64180bd24c403b444 Mon Sep 17 00:00:00 2001 From: Richard Lundeen Date: Wed, 15 Jul 2026 14:04:44 -0700 Subject: [PATCH 04/10] DOC: Clarify available as per-scenario pool-shaping hook distinct from membership Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: e0bc715e-cd3a-480c-afe8-5cac8339224f --- pyrit/registry/components/attack_technique_registry.py | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/pyrit/registry/components/attack_technique_registry.py b/pyrit/registry/components/attack_technique_registry.py index 275f4bf1b1..942d4f4f63 100644 --- a/pyrit/registry/components/attack_technique_registry.py +++ b/pyrit/registry/components/attack_technique_registry.py @@ -225,6 +225,14 @@ def build_technique_class_from_factories( omitted. An ``ALL`` aggregate is always added. available (TagQuery | None): Query selecting which of ``factories`` are available for this scenario (the pool). ``None`` means all of them. + This is a per-scenario *pool-shaping* hook, distinct from catalog + *membership* (which techniques exist at all) — membership is owned by + the active initializer, while ``available`` lets a single scenario + surface only a subset of the registered catalog. Use it for a + compound exclusion that reads better as one query than a manual + filter (e.g. ``TagQuery.none_of("foobar")`` to hide a + technique this scenario is incompatible with while it stays + registered for others). default (TagQuery | None): Query selecting the pool techniques that form the ``DEFAULT`` aggregate. Combined (union) with ``default_technique_names``. From 9867309365637ae1be6e7454d922741e2e829a8c Mon Sep 17 00:00:00 2001 From: Richard Lundeen Date: Wed, 15 Jul 2026 16:06:43 -0700 Subject: [PATCH 05/10] FIX: Default RapidResponse to light and drop benchmark's orphaned default set RapidResponse now defaults to the `light` aggregate (with an `all` fallback when no light-tagged factory survives the initializer's pool), matching AdversarialBenchmark. Removes the orphaned `default_technique_names` from both scenarios so `light` is the single, unambiguous curated default rather than a dead `default` aggregate nothing referenced. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: e0bc715e-cd3a-480c-afe8-5cac8339224f --- .../scenario/scenarios/airt/rapid_response.py | 9 ++++-- .../scenarios/benchmark/adversarial.py | 6 ++-- .../unit/scenario/airt/test_rapid_response.py | 29 +++++++++++-------- 3 files changed, 26 insertions(+), 18 deletions(-) diff --git a/pyrit/scenario/scenarios/airt/rapid_response.py b/pyrit/scenario/scenarios/airt/rapid_response.py index 517fa6e934..43455f1127 100644 --- a/pyrit/scenario/scenarios/airt/rapid_response.py +++ b/pyrit/scenario/scenarios/airt/rapid_response.py @@ -50,7 +50,6 @@ def _build_rapid_response_technique() -> type[ScenarioTechnique]: return AttackTechniqueRegistry.build_technique_class_from_factories( # type: ignore[ty:invalid-return-type] class_name="RapidResponseTechnique", factories=factories, - default_technique_names={"role_play_movie_script", "many_shot"}, ) @@ -90,11 +89,17 @@ def __init__( technique_class = _build_rapid_response_technique() + # ``light`` is the curated default run, but it only exists when the pool + # (decided by the active initializer) contains a light-tagged factory. Fall + # back to ``all`` so a custom technique set with no light factory still yields + # a constructible scenario rather than raising on an absent aggregate. + default_aggregate = "light" if "light" in technique_class.get_aggregate_tags() else "all" + super().__init__( version=self.VERSION, objective_scorer=self._objective_scorer, technique_class=technique_class, - default_technique=technique_class("default"), + default_technique=technique_class(default_aggregate), default_dataset_config=CompoundDatasetAttackConfiguration.per_dataset( dataset_names=[ "airt_hate", diff --git a/pyrit/scenario/scenarios/benchmark/adversarial.py b/pyrit/scenario/scenarios/benchmark/adversarial.py index 4e9304e5cb..1c1af767b9 100644 --- a/pyrit/scenario/scenarios/benchmark/adversarial.py +++ b/pyrit/scenario/scenarios/benchmark/adversarial.py @@ -43,9 +43,8 @@ def _build_benchmark_technique() -> type[ScenarioTechnique]: decided by the active initializer (the registration gate); this scenario does not narrow the pool further by group. The resulting enum has one concrete member per factory (e.g. ``red_teaming``, ``tap``, - ``crescendo_simulated``), a ``light`` / ``single_turn`` / ``multi_turn`` - aggregate for each catalog tag, and a ``default`` aggregate for the - scenario's default run. + ``crescendo_simulated``) and a ``light`` / ``single_turn`` / ``multi_turn`` + aggregate for each catalog tag. ``light`` is the scenario's default run. The (technique × target) cross-product is materialized lazily in ``AdversarialBenchmark._build_atomic_attacks_async`` from the @@ -63,7 +62,6 @@ def _build_benchmark_technique() -> type[ScenarioTechnique]: return AttackTechniqueRegistry.build_technique_class_from_factories( # type: ignore[ty:invalid-return-type] class_name="BenchmarkTechnique", factories=factories, - default_technique_names={"role_play_movie_script", "many_shot"}, ) diff --git a/tests/unit/scenario/airt/test_rapid_response.py b/tests/unit/scenario/airt/test_rapid_response.py index 29d8f4653c..66a7d1be35 100644 --- a/tests/unit/scenario/airt/test_rapid_response.py +++ b/tests/unit/scenario/airt/test_rapid_response.py @@ -165,12 +165,12 @@ def test_get_technique_class(self, mock_objective_scorer): ): assert RapidResponse()._technique_class is strat - def test_get_default_technique_returns_default(self, mock_objective_scorer): + def test_get_default_technique_returns_light(self, mock_objective_scorer): strat = _technique_class() with patch( "pyrit.scenario.core.scenario.Scenario._get_default_objective_scorer", return_value=mock_objective_scorer ): - assert RapidResponse()._default_technique == strat.DEFAULT + assert RapidResponse()._default_technique == strat.LIGHT def test_default_dataset_config_has_all_harm_datasets(self, mock_objective_scorer): with patch( @@ -210,7 +210,7 @@ def test_initialization_with_custom_scorer(self, mock_objective_scorer): new_callable=AsyncMock, return_value=ALL_HARM_SEED_GROUPS, ) - async def test_initialization_defaults_to_default_technique( + async def test_initialization_defaults_to_light_technique( self, _mock_groups, mock_get_scorer, @@ -221,8 +221,11 @@ async def test_initialization_defaults_to_default_technique( scenario = RapidResponse() scenario.set_params_from_args(args={"objective_target": mock_objective_target}) await scenario.initialize_async() - # DEFAULT expands to PromptSending + ManyShot → 2 composites - assert len(scenario._scenario_techniques) == 2 + # Default is the LIGHT aggregate; it expands to every light-tagged technique. + strat = _technique_class() + expected = len(strat.expand({strat.LIGHT})) + assert expected > 2 + assert len(scenario._scenario_techniques) == expected async def test_initialize_raises_when_no_datasets(self, mock_objective_target, mock_objective_scorer): """Dataset resolution fails from empty memory.""" @@ -306,16 +309,17 @@ async def _init_and_get_attacks( await scenario.initialize_async() return scenario._atomic_attacks - async def test_default_technique_produces_role_play_and_many_shot( - self, mock_objective_target, mock_objective_scorer - ): + async def test_default_technique_is_light(self, mock_objective_target, mock_objective_scorer): attacks = await self._init_and_get_attacks( mock_objective_target=mock_objective_target, mock_objective_scorer=mock_objective_scorer, ) technique_classes = {type(a.attack_technique.attack) for a in attacks} - # role_play_movie_script is now a simulated-conversation PromptSendingAttack. + # Default is the LIGHT aggregate: it includes many_shot, context_compliance, and the + # simulated-conversation role_play_* variants, but excludes the slow TAP attack. assert ManyShotJailbreakAttack in technique_classes + assert ContextComplianceAttack in technique_classes + assert TreeOfAttacksWithPruningAttack not in technique_classes assert any( a.attack_technique.seed_technique is not None and a.attack_technique.seed_technique.has_simulated_conversation @@ -432,7 +436,7 @@ def _spy_create(self, **kwargs): assert len(extra) == 1 async def test_attack_count_is_techniques_times_datasets(self, mock_objective_target, mock_objective_scorer): - """With 2 datasets and DEFAULT (2 techniques), expect 4 atomic attacks.""" + """With 2 datasets and the LIGHT default, expect (light techniques) x 2 atomic attacks.""" two_datasets = { "hate": _make_seed_groups("hate"), "violence": _make_seed_groups("violence"), @@ -442,8 +446,9 @@ async def test_attack_count_is_techniques_times_datasets(self, mock_objective_ta mock_objective_scorer=mock_objective_scorer, seed_groups=two_datasets, ) - # DEFAULT = RolePlay + ManyShot = 2 techniques, 2 datasets → 4 - assert len(attacks) == 4 + strat = _technique_class() + expected = len(strat.expand({strat.LIGHT})) * 2 + assert len(attacks) == expected async def test_atomic_attack_names_are_unique_compound_keys(self, mock_objective_target, mock_objective_scorer): """Each AtomicAttack has a unique compound atomic_attack_name for resume correctness.""" From dc5e6b985f7c401db655370ca35faf485fe7cb41 Mon Sep 17 00:00:00 2001 From: Richard Lundeen Date: Wed, 15 Jul 2026 16:49:27 -0700 Subject: [PATCH 06/10] REFACTOR: Move default technique ownership into the technique catalog Scenarios previously passed default_technique to the base Scenario constructor, duplicating the light->all fallback in rapid_response and adversarial benchmark __init__. Move that ownership onto the technique class: ScenarioTechnique.default() resolves the catalog default, the builder records it once (with the light->all fallback), and the base Scenario reads it via technique_class.default(). Static-enum scenarios override default() with their member. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: e0bc715e-cd3a-480c-afe8-5cac8339224f --- .../components/attack_technique_registry.py | 58 ++++++++++++------- pyrit/scenario/core/scenario.py | 9 ++- pyrit/scenario/core/scenario_technique.py | 20 +++++++ .../scenarios/adaptive/adaptive_scenario.py | 7 --- .../scenarios/adaptive/text_adaptive.py | 6 -- pyrit/scenario/scenarios/airt/cyber.py | 1 - pyrit/scenario/scenarios/airt/jailbreak.py | 6 +- pyrit/scenario/scenarios/airt/leakage.py | 1 - pyrit/scenario/scenarios/airt/psychosocial.py | 1 - .../scenario/scenarios/airt/rapid_response.py | 8 +-- pyrit/scenario/scenarios/airt/scam.py | 6 +- .../scenarios/benchmark/adversarial.py | 8 +-- .../scenarios/foundry/red_team_agent.py | 6 +- pyrit/scenario/scenarios/garak/doctor.py | 1 - pyrit/scenario/scenarios/garak/encoding.py | 1 - .../scenario/scenarios/garak/web_injection.py | 6 +- tests/unit/scenario/core/test_scenario.py | 2 - .../scenario/core/test_scenario_parameters.py | 1 - .../core/test_scenario_partial_results.py | 1 - .../unit/scenario/core/test_scenario_retry.py | 1 - .../scenarios/adaptive/test_text_adaptive.py | 3 +- 21 files changed, 84 insertions(+), 69 deletions(-) diff --git a/pyrit/registry/components/attack_technique_registry.py b/pyrit/registry/components/attack_technique_registry.py index 942d4f4f63..5590e0d4f1 100644 --- a/pyrit/registry/components/attack_technique_registry.py +++ b/pyrit/registry/components/attack_technique_registry.py @@ -179,7 +179,7 @@ def build_technique_class_from_factories( factories: list[AttackTechniqueFactory], aggregate_tags: dict[str, TagQuery] | None = None, available: TagQuery | None = None, - default: TagQuery | None = None, + default: str | None = None, default_technique_names: set[str] | None = None, ) -> type: """ @@ -204,14 +204,14 @@ def build_technique_class_from_factories( pool becomes one automatically; ``aggregate_tags`` adds explicit ones for queries a single tag can't express. Each is evaluated only over the pool, so every aggregate is a subset of available. - - **default**: what runs when the caller selects nothing. Given as a - ``TagQuery`` (``default``) and/or explicit ``default_technique_names``; - both are intersected with the pool, so ``DEFAULT`` is always a subset of - available. + - **default**: what runs when the caller selects nothing. ``default`` names + the member (an aggregate or a single technique) the catalog treats as its + default; it is recorded on the class and returned by + ``ScenarioTechnique.default()``. When omitted, a ``DEFAULT`` aggregate + (built from ``default_technique_names``) is used if present, else ``ALL``. - ``default`` is chosen per-scenario: a scenario picks its default set by query - or by name, so the same technique can be the default for one scenario and not - another. + The default is chosen per-scenario, so the same technique can be the default for + one scenario and not another. Args: class_name (str): Name for the generated enum class. @@ -233,15 +233,16 @@ def build_technique_class_from_factories( filter (e.g. ``TagQuery.none_of("foobar")`` to hide a technique this scenario is incompatible with while it stays registered for others). - default (TagQuery | None): Query selecting the pool techniques that form - the ``DEFAULT`` aggregate. Combined (union) with - ``default_technique_names``. + default (str | None): Value of the member (aggregate or technique) the + catalog uses as its default. Falls back to ``all`` when that member is + absent from the pool (e.g. a custom initializer registers no + ``light``-tagged factory). When ``None``, the default is the ``DEFAULT`` + aggregate if one was built, otherwise ``all``. default_technique_names (set[str] | None): Names of pool techniques that - form this scenario's ``DEFAULT`` aggregate. Combined (union) with - ``default``. Names not present in the pool are ignored, so a scenario - can list its intended default set even when some of those techniques - are filtered out of its pool. When the combined default selection is - empty, no ``DEFAULT`` aggregate is generated. + form this scenario's ``DEFAULT`` aggregate. Names not present in the pool + are ignored, so a scenario can list its intended default set even when + some of those techniques are filtered out of its pool. When empty, no + ``DEFAULT`` aggregate is generated. Returns: type: A ``ScenarioTechnique`` subclass with the generated members. @@ -253,13 +254,12 @@ def build_technique_class_from_factories( # available (the pool): filter the candidate factories by the availability query. pool = available.filter(factories) if available is not None else list(factories) - # default: resolve from an explicit name set and/or a query over the pool. The - # DEFAULT aggregate exists whenever a default was requested; its membership is - # limited to the pool below (only pool factories are iterated), so DEFAULT is - # always a subset of available. + # default: the DEFAULT aggregate is built from ``default_technique_names``. Its + # membership is limited to the pool below (only pool factories are iterated), so + # DEFAULT is always a subset of available. Which member is the *catalog default* + # (used when the caller selects nothing) is recorded separately after the class + # is built, from the ``default`` argument. default_names: set[str] = set(default_technique_names or set()) - if default is not None: - default_names |= {f.name for f in pool if default.matches(set(f.technique_tags))} # Auto-promote every catalog tag present in the pool into a selectable # aggregate, so tags and aggregates are synonymous: selecting a tag expands to @@ -305,6 +305,20 @@ def _get_aggregate_tags(cls: type) -> set[str]: technique_cls.get_aggregate_tags = _get_aggregate_tags # type: ignore[ty:invalid-assignment] + # Record the catalog's default member (used when the caller selects nothing). + # ``default`` names an aggregate/technique to use; it falls back to ``all`` when + # that member is absent from the pool (e.g. a custom initializer registers no + # ``light``-tagged factory). Absent an explicit ``default``, a built ``DEFAULT`` + # aggregate is used when present, otherwise ``all``. + member_values = {member.value for member in technique_cls} + if default is not None and default in member_values: + default_value = default + elif default_names: + default_value = "default" + else: + default_value = "all" + technique_cls._default_technique_value = default_value # type: ignore[attr-defined] + return technique_cls # type: ignore[ty:invalid-return-type] def register_from_factories( diff --git a/pyrit/scenario/core/scenario.py b/pyrit/scenario/core/scenario.py index 7f5a9927c9..bbee5b09d3 100644 --- a/pyrit/scenario/core/scenario.py +++ b/pyrit/scenario/core/scenario.py @@ -153,7 +153,6 @@ def __init__( name: str = "", version: int, technique_class: type[ScenarioTechnique], - default_technique: ScenarioTechnique, default_dataset_config: DatasetAttackConfiguration, objective_scorer: Scorer, scenario_result_id: uuid.UUID | str | None = None, @@ -165,9 +164,9 @@ def __init__( name (str): Descriptive name for the scenario. version (int): Version number of the scenario. technique_class (type[ScenarioTechnique]): The technique enum class for this scenario. - default_technique (ScenarioTechnique): The default technique member used when no - ``scenario_techniques`` are passed to ``initialize_async``. Usually an aggregate - member like ``MyTechnique.ALL`` or ``MyTechnique.DEFAULT``. + The technique used when no ``scenario_techniques`` are passed to + ``initialize_async`` is the catalog's own default, read from + ``technique_class.default()``. default_dataset_config (DatasetAttackConfiguration): The default dataset configuration used when no ``dataset_config`` is passed to ``initialize_async``. objective_scorer (Scorer): The objective scorer used to evaluate attack results. @@ -197,7 +196,7 @@ def __init__( # Store technique configuration for use in initialize_async self._technique_class = technique_class - self._default_technique = default_technique + self._default_technique = technique_class.default() self._default_dataset_config = default_dataset_config # These will be set in initialize_async diff --git a/pyrit/scenario/core/scenario_technique.py b/pyrit/scenario/core/scenario_technique.py index 07550cc22f..dfcf1a8486 100644 --- a/pyrit/scenario/core/scenario_technique.py +++ b/pyrit/scenario/core/scenario_technique.py @@ -114,6 +114,26 @@ def tags(self) -> set[str]: """ return self._tags + @classmethod + def default(cls: type[T]) -> T: + """ + Return the technique member used when the caller selects nothing. + + The default is a property of the technique catalog, not of the scenario that + consumes it. Dynamically built classes have their default recorded by + ``build_technique_class_from_factories`` (stored as ``_default_technique_value``). + Statically declared subclasses whose default is not ``ALL`` override this + classmethod to return their default member; everything else falls back to ``ALL``, + which every catalog always contains. + + Returns: + T: The default technique member. + """ + value = getattr(cls, "_default_technique_value", None) + if value is not None: + return cls(value) + return cls["ALL"] + @classmethod def get_aggregate_tags(cls: type[T]) -> set[str]: """ diff --git a/pyrit/scenario/scenarios/adaptive/adaptive_scenario.py b/pyrit/scenario/scenarios/adaptive/adaptive_scenario.py index 5a9b812950..6916a7189e 100644 --- a/pyrit/scenario/scenarios/adaptive/adaptive_scenario.py +++ b/pyrit/scenario/scenarios/adaptive/adaptive_scenario.py @@ -71,12 +71,6 @@ def get_technique_class(cls) -> type[ScenarioTechnique]: """Return the scenario's technique enum (subclasses must override).""" raise NotImplementedError - @classmethod - @abstractmethod - def get_default_technique(cls) -> ScenarioTechnique: - """Return the scenario's default technique aggregate (subclasses must override).""" - raise NotImplementedError - @classmethod @abstractmethod def default_dataset_config(cls) -> DatasetAttackConfiguration: @@ -108,7 +102,6 @@ def __init__( super().__init__( version=self.VERSION, technique_class=self.get_technique_class(), - default_technique=self.get_default_technique(), default_dataset_config=self.default_dataset_config(), objective_scorer=objective_scorer, scenario_result_id=scenario_result_id, diff --git a/pyrit/scenario/scenarios/adaptive/text_adaptive.py b/pyrit/scenario/scenarios/adaptive/text_adaptive.py index 537fb7f096..6fba3591d2 100644 --- a/pyrit/scenario/scenarios/adaptive/text_adaptive.py +++ b/pyrit/scenario/scenarios/adaptive/text_adaptive.py @@ -99,12 +99,6 @@ def get_technique_class(cls) -> type[ScenarioTechnique]: cls._cached_technique_class = _build_text_adaptive_technique() return cls._cached_technique_class - @classmethod - def get_default_technique(cls) -> ScenarioTechnique: - """Return the default technique aggregate (resolves to every ``default``-tagged technique).""" - technique_class = cls.get_technique_class() - return technique_class("default") - @classmethod def required_datasets(cls) -> list[str]: """Return the dataset names this scenario expects when no override is provided.""" diff --git a/pyrit/scenario/scenarios/airt/cyber.py b/pyrit/scenario/scenarios/airt/cyber.py index a9c56508b5..46d63cdc82 100644 --- a/pyrit/scenario/scenarios/airt/cyber.py +++ b/pyrit/scenario/scenarios/airt/cyber.py @@ -106,7 +106,6 @@ def __init__( version=self.VERSION, objective_scorer=self._objective_scorer, technique_class=technique_class, - default_technique=technique_class("default"), default_dataset_config=DatasetAttackConfiguration(dataset_names=["airt_malware"], max_dataset_size=4), scenario_result_id=scenario_result_id, ) diff --git a/pyrit/scenario/scenarios/airt/jailbreak.py b/pyrit/scenario/scenarios/airt/jailbreak.py index 16157f0316..acd8ac356c 100644 --- a/pyrit/scenario/scenarios/airt/jailbreak.py +++ b/pyrit/scenario/scenarios/airt/jailbreak.py @@ -64,6 +64,11 @@ def get_aggregate_tags(cls) -> set[str]: # Include base class aggregates ("all") and add scenario-specific ones return super().get_aggregate_tags() | {"simple", "complex"} + @classmethod + def default(cls) -> "JailbreakTechnique": + """Return the default technique (``SIMPLE``) used when the caller selects nothing.""" + return cls.SIMPLE + class Jailbreak(Scenario): """ @@ -145,7 +150,6 @@ def __init__( super().__init__( version=self.VERSION, technique_class=JailbreakTechnique, - default_technique=JailbreakTechnique.SIMPLE, default_dataset_config=DatasetAttackConfiguration(dataset_names=["airt_harms"], max_dataset_size=4), objective_scorer=self._objective_scorer, scenario_result_id=scenario_result_id, diff --git a/pyrit/scenario/scenarios/airt/leakage.py b/pyrit/scenario/scenarios/airt/leakage.py index 492e775b84..6c39651aa8 100644 --- a/pyrit/scenario/scenarios/airt/leakage.py +++ b/pyrit/scenario/scenarios/airt/leakage.py @@ -119,7 +119,6 @@ def __init__( super().__init__( version=self.VERSION, technique_class=technique_class, - default_technique=technique_class("default"), default_dataset_config=DatasetAttackConfiguration(dataset_names=["airt_leakage"], max_dataset_size=4), objective_scorer=objective_scorer, scenario_result_id=scenario_result_id, diff --git a/pyrit/scenario/scenarios/airt/psychosocial.py b/pyrit/scenario/scenarios/airt/psychosocial.py index e7f919b552..d324b1a43b 100644 --- a/pyrit/scenario/scenarios/airt/psychosocial.py +++ b/pyrit/scenario/scenarios/airt/psychosocial.py @@ -224,7 +224,6 @@ def __init__( super().__init__( version=self.VERSION, technique_class=PsychosocialTechnique, - default_technique=PsychosocialTechnique.ALL, default_dataset_config=DatasetAttackConfiguration( dataset_names=["airt_imminent_crisis"], max_dataset_size=4 ), diff --git a/pyrit/scenario/scenarios/airt/rapid_response.py b/pyrit/scenario/scenarios/airt/rapid_response.py index 43455f1127..551d550b3d 100644 --- a/pyrit/scenario/scenarios/airt/rapid_response.py +++ b/pyrit/scenario/scenarios/airt/rapid_response.py @@ -50,6 +50,7 @@ def _build_rapid_response_technique() -> type[ScenarioTechnique]: return AttackTechniqueRegistry.build_technique_class_from_factories( # type: ignore[ty:invalid-return-type] class_name="RapidResponseTechnique", factories=factories, + default="light", ) @@ -89,17 +90,10 @@ def __init__( technique_class = _build_rapid_response_technique() - # ``light`` is the curated default run, but it only exists when the pool - # (decided by the active initializer) contains a light-tagged factory. Fall - # back to ``all`` so a custom technique set with no light factory still yields - # a constructible scenario rather than raising on an absent aggregate. - default_aggregate = "light" if "light" in technique_class.get_aggregate_tags() else "all" - super().__init__( version=self.VERSION, objective_scorer=self._objective_scorer, technique_class=technique_class, - default_technique=technique_class(default_aggregate), default_dataset_config=CompoundDatasetAttackConfiguration.per_dataset( dataset_names=[ "airt_hate", diff --git a/pyrit/scenario/scenarios/airt/scam.py b/pyrit/scenario/scenarios/airt/scam.py index dbc6cd48af..b8ec4be679 100644 --- a/pyrit/scenario/scenarios/airt/scam.py +++ b/pyrit/scenario/scenarios/airt/scam.py @@ -77,6 +77,11 @@ def get_aggregate_tags(cls) -> set[str]: # Include base class aggregates ("all") and add scenario-specific ones return super().get_aggregate_tags() | {"default", "single_turn", "multi_turn"} + @classmethod + def default(cls) -> "ScamTechnique": + """Return the default technique (``DEFAULT``) used when the caller selects nothing.""" + return cls.DEFAULT + class Scam(Scenario): """ @@ -147,7 +152,6 @@ def __init__( super().__init__( version=self.VERSION, technique_class=ScamTechnique, - default_technique=ScamTechnique.DEFAULT, default_dataset_config=DatasetAttackConfiguration(dataset_names=["airt_scams"], max_dataset_size=4), objective_scorer=objective_scorer, scenario_result_id=scenario_result_id, diff --git a/pyrit/scenario/scenarios/benchmark/adversarial.py b/pyrit/scenario/scenarios/benchmark/adversarial.py index 1c1af767b9..e322e74646 100644 --- a/pyrit/scenario/scenarios/benchmark/adversarial.py +++ b/pyrit/scenario/scenarios/benchmark/adversarial.py @@ -62,6 +62,7 @@ def _build_benchmark_technique() -> type[ScenarioTechnique]: return AttackTechniqueRegistry.build_technique_class_from_factories( # type: ignore[ty:invalid-return-type] class_name="BenchmarkTechnique", factories=factories, + default="light", ) @@ -175,17 +176,10 @@ def __init__( technique_class = _build_benchmark_technique() - # ``light`` is the curated default run, but it only exists when the pool - # (decided by the active initializer) contains a light-tagged factory. Fall - # back to ``all`` so a custom technique set with no light factory still yields - # a constructible benchmark rather than raising on an absent aggregate. - default_aggregate = "light" if "light" in technique_class.get_aggregate_tags() else "all" - super().__init__( version=self.VERSION, objective_scorer=self._objective_scorer, technique_class=technique_class, - default_technique=technique_class(default_aggregate), default_dataset_config=DatasetAttackConfiguration( dataset_names=["harmbench"], max_dataset_size=8, diff --git a/pyrit/scenario/scenarios/foundry/red_team_agent.py b/pyrit/scenario/scenarios/foundry/red_team_agent.py index cee05fbc31..d05b8e53fd 100644 --- a/pyrit/scenario/scenarios/foundry/red_team_agent.py +++ b/pyrit/scenario/scenarios/foundry/red_team_agent.py @@ -186,6 +186,11 @@ def get_aggregate_tags(cls) -> set[str]: # Include base class aggregates ("all") and add Foundry-specific ones return super().get_aggregate_tags() | {"easy", "moderate", "difficult", "converter", "attack"} + @classmethod + def default(cls) -> "FoundryTechnique": + """Return the default technique (``EASY``) used when the caller selects nothing.""" + return cls.EASY + class RedTeamAgent(Scenario): """ @@ -242,7 +247,6 @@ def __init__( super().__init__( version=self.VERSION, technique_class=FoundryTechnique, - default_technique=FoundryTechnique.EASY, default_dataset_config=DatasetAttackConfiguration(dataset_names=["harmbench"], max_dataset_size=4), objective_scorer=objective_scorer, scenario_result_id=scenario_result_id, diff --git a/pyrit/scenario/scenarios/garak/doctor.py b/pyrit/scenario/scenarios/garak/doctor.py index cd3a206a4f..896f6c1e0d 100644 --- a/pyrit/scenario/scenarios/garak/doctor.py +++ b/pyrit/scenario/scenarios/garak/doctor.py @@ -137,7 +137,6 @@ def __init__( super().__init__( version=self.VERSION, technique_class=technique_class, - default_technique=technique_class("default"), default_dataset_config=DatasetAttackConfiguration(dataset_names=["garak_doctor"]), objective_scorer=objective_scorer, scenario_result_id=scenario_result_id, diff --git a/pyrit/scenario/scenarios/garak/encoding.py b/pyrit/scenario/scenarios/garak/encoding.py index 34dcc8471b..b1a9246896 100644 --- a/pyrit/scenario/scenarios/garak/encoding.py +++ b/pyrit/scenario/scenarios/garak/encoding.py @@ -154,7 +154,6 @@ def __init__( super().__init__( version=self.VERSION, technique_class=EncodingTechnique, - default_technique=EncodingTechnique.ALL, default_dataset_config=CompoundDatasetAttackConfiguration( configurations=[ EncodingDatasetConfiguration(dataset_names=["garak_slur_terms_en"], max_dataset_size=3), diff --git a/pyrit/scenario/scenarios/garak/web_injection.py b/pyrit/scenario/scenarios/garak/web_injection.py index 962837a892..74c480c2ef 100644 --- a/pyrit/scenario/scenarios/garak/web_injection.py +++ b/pyrit/scenario/scenarios/garak/web_injection.py @@ -198,6 +198,11 @@ def get_aggregate_tags(cls) -> set[str]: """Return the tags that represent aggregate categories.""" return {"all", "default", "exfil", "xss"} + @classmethod + def default(cls) -> "WebInjectionTechnique": + """Return the default technique (``DEFAULT``) used when the caller selects nothing.""" + return cls.DEFAULT + class WebInjection(Scenario): """ @@ -260,7 +265,6 @@ def __init__( super().__init__( version=self.VERSION, technique_class=WebInjectionTechnique, - default_technique=WebInjectionTechnique.DEFAULT, default_dataset_config=DatasetAttackConfiguration( dataset_names=[ DATASET_EXAMPLE_DOMAINS, diff --git a/tests/unit/scenario/core/test_scenario.py b/tests/unit/scenario/core/test_scenario.py index 8bf9b368ae..b496c86718 100644 --- a/tests/unit/scenario/core/test_scenario.py +++ b/tests/unit/scenario/core/test_scenario.py @@ -156,7 +156,6 @@ def get_aggregate_tags(cls) -> set[str]: return {"all"} kwargs.setdefault("technique_class", TestTechnique) - kwargs.setdefault("default_technique", kwargs["technique_class"].ALL) kwargs.setdefault("default_dataset_config", DatasetConfiguration()) # Add a mock scorer if not provided @@ -772,7 +771,6 @@ def get_aggregate_tags(cls) -> set[str]: return {"all"} kwargs.setdefault("technique_class", TestTechnique) - kwargs.setdefault("default_technique", kwargs["technique_class"].ALL) kwargs.setdefault("default_dataset_config", DatasetConfiguration()) # Use TrueFalseScorer mock if not provided diff --git a/tests/unit/scenario/core/test_scenario_parameters.py b/tests/unit/scenario/core/test_scenario_parameters.py index b7512e4df3..98fe9e28cb 100644 --- a/tests/unit/scenario/core/test_scenario_parameters.py +++ b/tests/unit/scenario/core/test_scenario_parameters.py @@ -65,7 +65,6 @@ async def _build_atomic_attacks_async(self, *, context): return _ParamTestScenario( version=1, technique_class=_ParamTestTechnique, - default_technique=_ParamTestTechnique.ALL, default_dataset_config=DatasetConfiguration(), objective_scorer=mock_scorer, ) diff --git a/tests/unit/scenario/core/test_scenario_partial_results.py b/tests/unit/scenario/core/test_scenario_partial_results.py index 74542813fe..c95fa7f2e6 100644 --- a/tests/unit/scenario/core/test_scenario_partial_results.py +++ b/tests/unit/scenario/core/test_scenario_partial_results.py @@ -104,7 +104,6 @@ def __init__(self, *, atomic_attacks_to_return=None, objective_scorer=None, **kw objective_scorer = MagicMock() objective_scorer.get_identifier.return_value = _mock_scorer_id("MockScorer") - kwargs.setdefault("default_technique", technique_class.ALL) kwargs.setdefault("default_dataset_config", DatasetConfiguration()) super().__init__(technique_class=technique_class, objective_scorer=objective_scorer, **kwargs) self._test_atomic_attacks = atomic_attacks_to_return or [] diff --git a/tests/unit/scenario/core/test_scenario_retry.py b/tests/unit/scenario/core/test_scenario_retry.py index 64553dc1dc..efd3651939 100644 --- a/tests/unit/scenario/core/test_scenario_retry.py +++ b/tests/unit/scenario/core/test_scenario_retry.py @@ -175,7 +175,6 @@ def __init__(self, *, atomic_attacks_to_return=None, objective_scorer=None, **kw objective_scorer = MagicMock() objective_scorer.get_identifier.return_value = _mock_scorer_id("MockScorer") - kwargs.setdefault("default_technique", technique_class.ALL) kwargs.setdefault("default_dataset_config", DatasetConfiguration()) super().__init__(technique_class=technique_class, objective_scorer=objective_scorer, **kwargs) self._atomic_attacks_to_return = atomic_attacks_to_return or [] diff --git a/tests/unit/scenario/scenarios/adaptive/test_text_adaptive.py b/tests/unit/scenario/scenarios/adaptive/test_text_adaptive.py index 383d5e4fd3..3efc712c86 100644 --- a/tests/unit/scenario/scenarios/adaptive/test_text_adaptive.py +++ b/tests/unit/scenario/scenarios/adaptive/test_text_adaptive.py @@ -135,9 +135,10 @@ def test_get_technique_class_is_cached(self): assert cls_a is cls_b def test_get_default_technique(self): - strat = TextAdaptive.get_default_technique() + strat = TextAdaptive.get_technique_class().default() # The default aggregate must resolve to something runnable. assert strat is not None + assert strat.value == "default" @patch("pyrit.scenario.core.scenario.Scenario._get_default_objective_scorer") def test_init_stores_adaptive_params(self, mock_get_scorer, mock_objective_scorer): From f337d8313ffe8bad133d8f13c379a4668288865f Mon Sep 17 00:00:00 2001 From: Richard Lundeen Date: Wed, 15 Jul 2026 17:00:24 -0700 Subject: [PATCH 07/10] REFACTOR: Simplify build_technique_class_from_factories signature Drop two provably-dead parameters (no caller or test passed either) and reconcile the two default idioms into a clean either/or: - Remove aggregate_tags: catalog tags already auto-promote to selectable aggregates, so the explicit map was redundant. - Remove available: the caller already hands in factories and pre-filters the pool, so a second in-function filter just duplicated that. - Replace default (str) + default_technique_names (set) with mutually exclusive default_tags / default_names; both build the DEFAULT aggregate and raise if both are given. Also collapse the "no default -> ALL" fallback to a single owner: the builder now only records _default_technique_value when a DEFAULT aggregate was actually built, and ScenarioTechnique.default() owns the ALL fallback. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: e0bc715e-cd3a-480c-afe8-5cac8339224f --- doc/code/framework.md | 2 +- .../components/attack_technique_registry.py | 161 +++++++----------- .../scenarios/adaptive/text_adaptive.py | 2 +- pyrit/scenario/scenarios/airt/cyber.py | 2 +- pyrit/scenario/scenarios/airt/leakage.py | 2 +- .../scenario/scenarios/airt/rapid_response.py | 2 +- .../scenarios/benchmark/adversarial.py | 2 +- pyrit/scenario/scenarios/garak/doctor.py | 2 +- pyrit/setup/initializers/techniques/core.py | 2 +- .../unit/scenario/airt/test_rapid_response.py | 20 ++- 10 files changed, 82 insertions(+), 115 deletions(-) diff --git a/doc/code/framework.md b/doc/code/framework.md index 6bcb4a6890..ed95bcb63e 100644 --- a/doc/code/framework.md +++ b/doc/code/framework.md @@ -177,7 +177,7 @@ If you are contributing to PyRIT, that work will most likely land in one of the - Techniques are self-describing `AttackTechniqueFactory` instances (a `name`, `attack_class`, `attack_kwargs`, and `technique_tags`). They read like configuration even though they are code, so adding one — or bringing your own — is easy. The canonical catalog lives under `pyrit/setup/initializers/techniques/`: `core.py` (a small, curated standard set, auto-loaded on a bare `initialize_pyrit_async`), `extra.py` (the broader collection, opt-in), and per-source modules like `airt.py` (owned by a source/scenario but reusable, tagged with their owner and kept out of the default pool). - `core` stays deliberately small so a default run doesn't print 200 techniques or take forever; the wider catalog lives in `extra` and is selected on demand. Users pick subsets by passing initializer tags (e.g. `core`, `extra`, `all`) or writing their own initializer, so different runs — including from the CLI — can register different technique sets without changing the catalog. - A technique tied to one scenario is fine; if it's pinned and non-reusable it can stay local to that scenario, but if another scenario could reuse it, promote it to a catalog module and tag it. -- Tags describe a technique (behavioral tags like `single_turn`/`multi_turn`, owner tags like `airt`); they don't decide what a scenario runs. There is deliberately **no global `default` tag** — a default is scenario-relative, declared per scenario via `build_technique_class_from_factories` (`available` selects the pool, aggregates are named presets, `default` / `default_technique_names` set what runs when nothing is chosen). +- Tags describe a technique (behavioral tags like `single_turn`/`multi_turn`, owner tags like `airt`); they don't decide what a scenario runs. There is deliberately **no global `default` tag** — a default is scenario-relative, declared per scenario via `build_technique_class_from_factories` (the `factories` list is the pool, catalog tags become named aggregate presets, and `default_tags` / `default_names` set what runs when nothing is chosen). - **Does not own**: the conversation algorithm itself. Branching, turn management, and scoring decisions live in the executor it wraps — a technique only selects and configures existing components, and shouldn't implement new sending, scoring, or branching logic. **Framework Plans**: diff --git a/pyrit/registry/components/attack_technique_registry.py b/pyrit/registry/components/attack_technique_registry.py index 5590e0d4f1..dbcdf29958 100644 --- a/pyrit/registry/components/attack_technique_registry.py +++ b/pyrit/registry/components/attack_technique_registry.py @@ -29,7 +29,6 @@ from pyrit.registry.registry_metadata import RegistryMetadata if TYPE_CHECKING: - from pyrit.registry.tag_query import TagQuery from pyrit.scenario.core.attack_technique_factory import ( AttackTechniqueFactory, ScorerOverridePolicy, @@ -177,123 +176,95 @@ def build_technique_class_from_factories( *, class_name: str, factories: list[AttackTechniqueFactory], - aggregate_tags: dict[str, TagQuery] | None = None, - available: TagQuery | None = None, - default: str | None = None, - default_technique_names: set[str] | None = None, + default_tags: set[str] | None = None, + default_names: set[str] | None = None, ) -> type: """ Build a ``ScenarioTechnique`` enum subclass dynamically from technique factories. Creates an enum class with: - An ``ALL`` aggregate member (always included). - - A ``DEFAULT`` aggregate member when a default selection is provided. - - Additional aggregate members from ``aggregate_tags`` keys. - - An aggregate member for every other catalog tag present in the pool, so - tags and aggregates are synonymous: selecting a tag (e.g. ``core`` or a - custom ``airt_internal``) expands to every technique carrying it. - - One technique member per *available* factory, with tags from the factory. - - The three selection roles are all expressed the same way — as tag queries - over ``factories`` — and relate as strict subsets: - - - **available** (the pool): ``available`` filters ``factories`` to the - techniques this scenario exposes. When ``None`` the whole ``factories`` - list is the pool (back-compatible). - - **aggregates**: selectable tag groups. Every catalog tag present in the - pool becomes one automatically; ``aggregate_tags`` adds explicit ones for - queries a single tag can't express. Each is evaluated only over the pool, - so every aggregate is a subset of available. - - **default**: what runs when the caller selects nothing. ``default`` names - the member (an aggregate or a single technique) the catalog treats as its - default; it is recorded on the class and returned by - ``ScenarioTechnique.default()``. When omitted, a ``DEFAULT`` aggregate - (built from ``default_technique_names``) is used if present, else ``ALL``. - - The default is chosen per-scenario, so the same technique can be the default for - one scenario and not another. + - A ``DEFAULT`` aggregate member when a default selection is provided and at + least one pool technique matches it. + - An aggregate member for every catalog tag present in the pool, so tags and + aggregates are synonymous: selecting a tag (e.g. ``core`` or a custom + ``airt_internal``) expands to every technique carrying it. + - One technique member per factory, with tags from the factory. + + The catalog's *default* — what runs when the caller selects nothing — is defined + by exactly one of ``default_tags`` or ``default_names`` (or neither). Both build + the synthetic ``DEFAULT`` aggregate and are recorded on the class, returned by + ``ScenarioTechnique.default()``. When neither is given, or the chosen set matches + no pool technique (e.g. a custom initializer registers no ``light``-tagged + factory), the default falls back to ``ALL``. The default is chosen per-scenario, + so the same technique can be the default for one scenario and not another. Args: class_name (str): Name for the generated enum class. - factories (list[AttackTechniqueFactory]): Candidate technique factories. - Filtered by ``available`` to form the pool of enum members. - aggregate_tags (dict[str, TagQuery] | None): Optional explicit aggregate - definitions, mapping an aggregate member name to a ``TagQuery`` over the - pool. Only needed for aggregates a single tag can't express — a compound - query (AND/OR/NOT) or an aggregate whose name is not itself a tag. Simple - one-to-one tag aggregates are auto-derived (see above), so this can be - omitted. An ``ALL`` aggregate is always added. - available (TagQuery | None): Query selecting which of ``factories`` are - available for this scenario (the pool). ``None`` means all of them. - This is a per-scenario *pool-shaping* hook, distinct from catalog - *membership* (which techniques exist at all) — membership is owned by - the active initializer, while ``available`` lets a single scenario - surface only a subset of the registered catalog. Use it for a - compound exclusion that reads better as one query than a manual - filter (e.g. ``TagQuery.none_of("foobar")`` to hide a - technique this scenario is incompatible with while it stays - registered for others). - default (str | None): Value of the member (aggregate or technique) the - catalog uses as its default. Falls back to ``all`` when that member is - absent from the pool (e.g. a custom initializer registers no - ``light``-tagged factory). When ``None``, the default is the ``DEFAULT`` - aggregate if one was built, otherwise ``all``. - default_technique_names (set[str] | None): Names of pool techniques that - form this scenario's ``DEFAULT`` aggregate. Names not present in the pool - are ignored, so a scenario can list its intended default set even when - some of those techniques are filtered out of its pool. When empty, no - ``DEFAULT`` aggregate is generated. + factories (list[AttackTechniqueFactory]): The technique factories that form + this scenario's pool of enum members. Callers pre-filter this list to + shape the pool. + default_tags (set[str] | None): Tags whose union defines the scenario's + ``DEFAULT`` aggregate — every pool technique carrying any of these tags is + the default (e.g. ``{"light"}``). Mutually exclusive with ``default_names``. + default_names (set[str] | None): Exact technique names that form the + scenario's ``DEFAULT`` aggregate. Names not present in the pool are + ignored, so a scenario can list its intended default set even when some of + those techniques are filtered out. Mutually exclusive with ``default_tags``. Returns: type: A ``ScenarioTechnique`` subclass with the generated members. + + Raises: + ValueError: If both ``default_tags`` and ``default_names`` are provided. """ from pyrit.scenario import ScenarioTechnique - aggregate_tags = aggregate_tags or {} - - # available (the pool): filter the candidate factories by the availability query. - pool = available.filter(factories) if available is not None else list(factories) + if default_tags and default_names: + raise ValueError("Provide at most one of default_tags or default_names, not both.") - # default: the DEFAULT aggregate is built from ``default_technique_names``. Its - # membership is limited to the pool below (only pool factories are iterated), so - # DEFAULT is always a subset of available. Which member is the *catalog default* - # (used when the caller selects nothing) is recorded separately after the class - # is built, from the ``default`` argument. - default_names: set[str] = set(default_technique_names or set()) - - # Auto-promote every catalog tag present in the pool into a selectable - # aggregate, so tags and aggregates are synonymous: selecting a tag expands to - # every technique carrying it. "all" and "default" are reserved synthetic - # aggregates (see above) and are never derived from tags. A tag that collides - # with a technique name stays a concrete technique (name selection wins), and - # tags already declared in aggregate_tags keep their explicit TagQuery. - reserved_aggregate_tags = {"all", "default"} + pool = list(factories) pool_technique_names = {f.name for f in pool} pool_tags = {tag for f in pool for tag in f.technique_tags} - auto_aggregate_tags = pool_tags - reserved_aggregate_tags - pool_technique_names - set(aggregate_tags) - all_aggregate_tag_names = {"all"} | set(aggregate_tags.keys()) | auto_aggregate_tags + # default: the pool techniques that form the DEFAULT aggregate, from either an + # explicit set of names or the union over a set of tags. Limited to the pool, so + # DEFAULT is always a subset of ALL. When it is empty (nothing matched) no DEFAULT + # aggregate is built and the catalog default falls back to ALL. if default_names: + default_member_names = {f.name for f in pool if f.name in default_names} + elif default_tags: + default_member_names = {f.name for f in pool if set(f.technique_tags) & default_tags} + else: + default_member_names = set() + + # Auto-promote every catalog tag present in the pool into a selectable aggregate, + # so tags and aggregates are synonymous: selecting a tag expands to every technique + # carrying it. "all" and "default" are reserved synthetic aggregates and are never + # derived from tags. A tag that collides with a technique name stays a concrete + # technique (name selection wins). + reserved_aggregate_tags = {"all", "default"} + auto_aggregate_tags = pool_tags - reserved_aggregate_tags - pool_technique_names + + all_aggregate_tag_names = {"all"} | auto_aggregate_tags + if default_member_names: all_aggregate_tag_names.add("default") members: dict[str, tuple[str, set[str]]] = {} # Aggregate members first (ALL is always present) members["ALL"] = ("all", {"all"}) - if default_names: + if default_member_names: members["DEFAULT"] = ("default", {"default"}) - for agg_name in aggregate_tags: - members[agg_name.upper()] = (agg_name, {agg_name}) for agg_name in sorted(auto_aggregate_tags): members[agg_name.upper()] = (agg_name, {agg_name}) - # Technique members from the pool — assign aggregate tags based on TagQuery matching + # Technique members from the pool — tag DEFAULT members so the aggregate expands. for factory in pool: factory_tags = set(factory.technique_tags) - matched_agg_tags = {agg_name for agg_name, query in aggregate_tags.items() if query.matches(factory_tags)} - if factory.name in default_names: - matched_agg_tags.add("default") - members[factory.name] = (factory.name, factory_tags | matched_agg_tags) + if factory.name in default_member_names: + factory_tags = factory_tags | {"default"} + members[factory.name] = (factory.name, factory_tags) # Build the enum class dynamically technique_cls = ScenarioTechnique(class_name, members) @@ -305,19 +276,11 @@ def _get_aggregate_tags(cls: type) -> set[str]: technique_cls.get_aggregate_tags = _get_aggregate_tags # type: ignore[ty:invalid-assignment] - # Record the catalog's default member (used when the caller selects nothing). - # ``default`` names an aggregate/technique to use; it falls back to ``all`` when - # that member is absent from the pool (e.g. a custom initializer registers no - # ``light``-tagged factory). Absent an explicit ``default``, a built ``DEFAULT`` - # aggregate is used when present, otherwise ``all``. - member_values = {member.value for member in technique_cls} - if default is not None and default in member_values: - default_value = default - elif default_names: - default_value = "default" - else: - default_value = "all" - technique_cls._default_technique_value = default_value # type: ignore[attr-defined] + # Record the catalog's default only when a DEFAULT aggregate was actually built. + # When it wasn't, the attribute is left unset and ScenarioTechnique.default() owns + # the single ALL fallback — so the "no default -> ALL" rule lives in one place. + if default_member_names: + technique_cls._default_technique_value = "default" # type: ignore[attr-defined] return technique_cls # type: ignore[ty:invalid-return-type] diff --git a/pyrit/scenario/scenarios/adaptive/text_adaptive.py b/pyrit/scenario/scenarios/adaptive/text_adaptive.py index 6fba3591d2..ff31ed29d8 100644 --- a/pyrit/scenario/scenarios/adaptive/text_adaptive.py +++ b/pyrit/scenario/scenarios/adaptive/text_adaptive.py @@ -70,7 +70,7 @@ def _build_text_adaptive_technique() -> type[ScenarioTechnique]: return AttackTechniqueRegistry.build_technique_class_from_factories( # type: ignore[return-value, ty:invalid-return-type] class_name="TextAdaptiveTechnique", factories=factories, - default_technique_names={"role_play_movie_script", "many_shot"}, + default_names={"role_play_movie_script", "many_shot"}, ) diff --git a/pyrit/scenario/scenarios/airt/cyber.py b/pyrit/scenario/scenarios/airt/cyber.py index 46d63cdc82..22022bba0c 100644 --- a/pyrit/scenario/scenarios/airt/cyber.py +++ b/pyrit/scenario/scenarios/airt/cyber.py @@ -53,7 +53,7 @@ def _build_cyber_technique() -> type[ScenarioTechnique]: return AttackTechniqueRegistry.build_technique_class_from_factories( # type: ignore[ty:invalid-return-type] class_name="CyberTechnique", factories=factories, - default_technique_names=_CYBER_DEFAULT_TECHNIQUE_NAMES, + default_names=_CYBER_DEFAULT_TECHNIQUE_NAMES, ) diff --git a/pyrit/scenario/scenarios/airt/leakage.py b/pyrit/scenario/scenarios/airt/leakage.py index 6c39651aa8..7fb356cbc8 100644 --- a/pyrit/scenario/scenarios/airt/leakage.py +++ b/pyrit/scenario/scenarios/airt/leakage.py @@ -66,7 +66,7 @@ def _build_leakage_technique() -> type[ScenarioTechnique]: return AttackTechniqueRegistry.build_technique_class_from_factories( # type: ignore[return-value, ty:invalid-return-type] class_name="LeakageTechnique", factories=all_factories, - default_technique_names={"role_play_movie_script", "many_shot", "first_letter", "image"}, + default_names={"role_play_movie_script", "many_shot", "first_letter", "image"}, ) diff --git a/pyrit/scenario/scenarios/airt/rapid_response.py b/pyrit/scenario/scenarios/airt/rapid_response.py index 551d550b3d..c981657c48 100644 --- a/pyrit/scenario/scenarios/airt/rapid_response.py +++ b/pyrit/scenario/scenarios/airt/rapid_response.py @@ -50,7 +50,7 @@ def _build_rapid_response_technique() -> type[ScenarioTechnique]: return AttackTechniqueRegistry.build_technique_class_from_factories( # type: ignore[ty:invalid-return-type] class_name="RapidResponseTechnique", factories=factories, - default="light", + default_tags={"light"}, ) diff --git a/pyrit/scenario/scenarios/benchmark/adversarial.py b/pyrit/scenario/scenarios/benchmark/adversarial.py index e322e74646..c6b3201add 100644 --- a/pyrit/scenario/scenarios/benchmark/adversarial.py +++ b/pyrit/scenario/scenarios/benchmark/adversarial.py @@ -62,7 +62,7 @@ def _build_benchmark_technique() -> type[ScenarioTechnique]: return AttackTechniqueRegistry.build_technique_class_from_factories( # type: ignore[ty:invalid-return-type] class_name="BenchmarkTechnique", factories=factories, - default="light", + default_tags={"light"}, ) diff --git a/pyrit/scenario/scenarios/garak/doctor.py b/pyrit/scenario/scenarios/garak/doctor.py index 896f6c1e0d..5daa89c95b 100644 --- a/pyrit/scenario/scenarios/garak/doctor.py +++ b/pyrit/scenario/scenarios/garak/doctor.py @@ -82,7 +82,7 @@ def _build_doctor_technique() -> type[ScenarioTechnique]: return AttackTechniqueRegistry.build_technique_class_from_factories( # type: ignore[return-value, ty:invalid-return-type] class_name="DoctorTechnique", factories=DOCTOR_FACTORIES, - default_technique_names={"policy_puppetry", "policy_puppetry_leet"}, + default_names={"policy_puppetry", "policy_puppetry_leet"}, ) diff --git a/pyrit/setup/initializers/techniques/core.py b/pyrit/setup/initializers/techniques/core.py index 5f6dda418b..1c7f9caee7 100644 --- a/pyrit/setup/initializers/techniques/core.py +++ b/pyrit/setup/initializers/techniques/core.py @@ -12,7 +12,7 @@ ``default`` is intentionally not a tag here: what runs by default is scenario-relative and is declared per scenario (see ``AttackTechniqueRegistry.build_technique_class_from_factories``'s -``default_technique_names``), not baked into the shared catalog. +``default_tags`` / ``default_names``), not baked into the shared catalog. """ from pathlib import Path diff --git a/tests/unit/scenario/airt/test_rapid_response.py b/tests/unit/scenario/airt/test_rapid_response.py index 66a7d1be35..65133ef66d 100644 --- a/tests/unit/scenario/airt/test_rapid_response.py +++ b/tests/unit/scenario/airt/test_rapid_response.py @@ -165,12 +165,15 @@ def test_get_technique_class(self, mock_objective_scorer): ): assert RapidResponse()._technique_class is strat - def test_get_default_technique_returns_light(self, mock_objective_scorer): + def test_get_default_technique_returns_default(self, mock_objective_scorer): strat = _technique_class() with patch( "pyrit.scenario.core.scenario.Scenario._get_default_objective_scorer", return_value=mock_objective_scorer ): - assert RapidResponse()._default_technique == strat.LIGHT + default = RapidResponse()._default_technique + assert default == strat.DEFAULT + # DEFAULT is built from default_tags={"light"}, so it expands to the light-tagged techniques. + assert strat.expand({strat.DEFAULT}) == strat.expand({strat.LIGHT}) def test_default_dataset_config_has_all_harm_datasets(self, mock_objective_scorer): with patch( @@ -221,9 +224,9 @@ async def test_initialization_defaults_to_light_technique( scenario = RapidResponse() scenario.set_params_from_args(args={"objective_target": mock_objective_target}) await scenario.initialize_async() - # Default is the LIGHT aggregate; it expands to every light-tagged technique. + # Default is the DEFAULT aggregate (built from light tags); it expands to every light-tagged technique. strat = _technique_class() - expected = len(strat.expand({strat.LIGHT})) + expected = len(strat.expand({strat.DEFAULT})) assert expected > 2 assert len(scenario._scenario_techniques) == expected @@ -315,8 +318,9 @@ async def test_default_technique_is_light(self, mock_objective_target, mock_obje mock_objective_scorer=mock_objective_scorer, ) technique_classes = {type(a.attack_technique.attack) for a in attacks} - # Default is the LIGHT aggregate: it includes many_shot, context_compliance, and the - # simulated-conversation role_play_* variants, but excludes the slow TAP attack. + # Default is the DEFAULT aggregate (built from light tags): it includes many_shot, + # context_compliance, and the simulated-conversation role_play_* variants, but excludes + # the slow TAP attack. assert ManyShotJailbreakAttack in technique_classes assert ContextComplianceAttack in technique_classes assert TreeOfAttacksWithPruningAttack not in technique_classes @@ -436,7 +440,7 @@ def _spy_create(self, **kwargs): assert len(extra) == 1 async def test_attack_count_is_techniques_times_datasets(self, mock_objective_target, mock_objective_scorer): - """With 2 datasets and the LIGHT default, expect (light techniques) x 2 atomic attacks.""" + """With 2 datasets and the DEFAULT (light) default, expect (light techniques) x 2 atomic attacks.""" two_datasets = { "hate": _make_seed_groups("hate"), "violence": _make_seed_groups("violence"), @@ -447,7 +451,7 @@ async def test_attack_count_is_techniques_times_datasets(self, mock_objective_ta seed_groups=two_datasets, ) strat = _technique_class() - expected = len(strat.expand({strat.LIGHT})) * 2 + expected = len(strat.expand({strat.DEFAULT})) * 2 assert len(attacks) == expected async def test_atomic_attack_names_are_unique_compound_keys(self, mock_objective_target, mock_objective_scorer): From 3d28bf895501f4628ab92661d450756d9676b00c Mon Sep 17 00:00:00 2001 From: Richard Lundeen Date: Wed, 15 Jul 2026 17:33:06 -0700 Subject: [PATCH 08/10] DOC: Document catalog-owned default technique + add builder tests Update scenarios.instructions.md to reflect that the default technique is owned by the technique catalog (technique_class.default()) rather than passed to super().__init__(): remove the stale default_technique super-arg from the code examples and requirements, add a "Default Technique" subsection covering the default() override for static enums and default_tags/default_names for dynamically built enums, and fix the stale TagQuery reference to describe auto-promoted aggregates. Add TestBuildTechniqueClassFromFactories covering tag auto-promotion, default_tags/default_names building the DEFAULT aggregate, the single-source ALL fallback when no default is declared (or nothing matches), and the ValueError when both defaults are passed. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: e0bc715e-cd3a-480c-afe8-5cac8339224f --- .../instructions/scenarios.instructions.md | 40 ++++++++-- .../test_attack_technique_registry.py | 79 ++++++++++++++++++- 2 files changed, 109 insertions(+), 10 deletions(-) diff --git a/.github/instructions/scenarios.instructions.md b/.github/instructions/scenarios.instructions.md index ee12e48750..71611d8df1 100644 --- a/.github/instructions/scenarios.instructions.md +++ b/.github/instructions/scenarios.instructions.md @@ -16,7 +16,7 @@ All scenarios inherit from `Scenario` (ABC) and must: 2. **Optionally declare `BASELINE_ATTACK_POLICY`** (defaults to `BaselineAttackPolicy.Enabled` — a baseline `PromptSendingAttack` is prepended and callers can opt out per run by setting `include_baseline=False` in the run params, see "Run Parameters" below): - `BaselineAttackPolicy.Disabled` — baseline supported but off by default (e.g. `Jailbreak`, where templates dominate the run). - `BaselineAttackPolicy.Forbidden` — baseline is meaningless for this scenario's comparison axis (e.g. `AdversarialBenchmark`, which compares against gold-standard answers). Supplying `include_baseline=True` raises `ValueError`. -3. **Pass `technique_class`, `default_technique`, and `default_dataset_config` to `super().__init__()`:** +3. **Pass `technique_class` and `default_dataset_config` to `super().__init__()`:** ```python class MyScenario(Scenario): @@ -28,13 +28,16 @@ class MyScenario(Scenario): super().__init__( version=self.VERSION, technique_class=MyTechnique, - default_technique=MyTechnique.ALL, default_dataset_config=DatasetConfiguration(dataset_names=["my_dataset"]), objective_scorer=objective_scorer or self._get_default_objective_scorer(), scenario_result_id=scenario_result_id, ) ``` +The default technique (what runs when the caller passes no `scenario_techniques`) is +**not** a constructor argument — it is owned by the technique catalog and read from +`technique_class.default()`. See "Default Technique" below. + For scenarios whose technique enum is built dynamically (RapidResponse pattern), build the technique class in a module-level `@cache`-decorated function and pass the result through the constructor — no classmethod indirection required. @@ -65,7 +68,6 @@ def __init__( super().__init__( version=self.VERSION, technique_class=MyTechnique, - default_technique=MyTechnique.ALL, default_dataset_config=DatasetConfiguration(dataset_names=["my_dataset"]), objective_scorer=objective_scorer, ) @@ -78,7 +80,7 @@ Requirements: `pyrit/common/brick_contract.py`). Violators raise `TypeError` at import time. - **All constructor parameters must be optional** (default to `None`) so the registry can instantiate the scenario with no arguments for metadata introspection. Defer required-input validation to `initialize_async()` or `_build_atomic_attacks_async()`. `ScenarioRegistry._build_metadata` raises `TypeError` if `scenario_class()` cannot be called with no arguments. -- `super().__init__()` called with `version`, `technique_class`, `default_technique`, `default_dataset_config`, `objective_scorer` +- `super().__init__()` called with `version`, `technique_class`, `default_dataset_config`, `objective_scorer` (the default technique is owned by the catalog, not passed here — see "Default Technique") - complex objects like `adversarial_chat` or `objective_scorer` should be passed into the constructor. ## Run Parameters @@ -153,12 +155,34 @@ class MyTechnique(ScenarioTechnique): @classmethod def get_aggregate_tags(cls) -> set[str]: return {"all", "default", "single_turn", "multi_turn"} + + @classmethod + def default(cls) -> "MyTechnique": + return cls.DEFAULT # what runs when the caller selects nothing ``` - `ALL` aggregate is always required - Each member: `NAME = ("string_value", {tag_set})` - Aggregates expand to all techniques matching their tag +### Default Technique + +The default technique — what runs when no `scenario_techniques` are passed — is a property +of the **technique catalog**, not the scenario. It is *not* passed to `super().__init__()`; +the base `Scenario` reads it from `technique_class.default()`. + +- **Statically declared enums:** override the `default()` classmethod to return the default + member (e.g. `return cls.DEFAULT`, `return cls.EASY`). If the default is `ALL`, don't + override — the base `ScenarioTechnique.default()` already falls back to `ALL`. +- **Dynamically built enums** (`build_technique_class_from_factories`, the RapidResponse + pattern): declare the default via exactly one of `default_tags={...}` (every pool technique + carrying any of those tags) or `default_names={...}` (exact technique names). Both build the + synthetic `DEFAULT` aggregate and record it so `default()` returns it. Passing both raises + `ValueError`. When neither is given (or nothing matches), the default falls back to `ALL`. + +The default is scenario-relative: the same catalog technique can be the default for one +scenario and not another, so there is deliberately no catalog-wide `default` tag. + ### Result grouping (`display_group`) `display_group` controls how attack results are aggregated for display. It is set per @@ -248,9 +272,11 @@ AttackTechniqueFactory( Key points: - `name` is required and must match the technique enum value the scenario looks up. -- `technique_tags` on the factory drives `TagQuery` filters used by - `AttackTechniqueRegistry.build_technique_class_from_factories(...)`. This is **distinct** - from the per-entry `tags` argument passed to `registry.register_technique(...)`. +- `technique_tags` on the factory drives the aggregates and default selection built by + `AttackTechniqueRegistry.build_technique_class_from_factories(...)`: every tag present in + the pool auto-promotes to a selectable aggregate, and `default_tags` picks the default set + by tag. This is **distinct** from the per-entry `tags` argument passed to + `registry.register_technique(...)`. - `uses_adversarial` is auto-derived from the attack class signature (presence of `attack_adversarial_config`) and seed shape; pass `False` explicitly to opt out, or `True` to force opt-in. diff --git a/tests/unit/registry/test_attack_technique_registry.py b/tests/unit/registry/test_attack_technique_registry.py index d0a109d365..cb9b7a64a1 100644 --- a/tests/unit/registry/test_attack_technique_registry.py +++ b/tests/unit/registry/test_attack_technique_registry.py @@ -303,9 +303,9 @@ def test_factory_attack_class_set(self, factory: AttackTechniqueFactory): def test_factory_attack_class_accepts_objective_target(self, factory: AttackTechniqueFactory): """Every attack class must accept ``objective_target`` (required at create time).""" sig = inspect.signature(factory.attack_class.__init__) - assert "objective_target" in sig.parameters, ( - f"{factory.attack_class.__name__} is missing required 'objective_target' parameter" - ) + assert ( + "objective_target" in sig.parameters + ), f"{factory.attack_class.__name__} is missing required 'objective_target' parameter" def test_factory_names_are_unique(self): """No two factories should share the same name.""" @@ -578,3 +578,76 @@ def test_factory_skips_silently_when_attack_has_no_scoring_param_and_policy_skip assert technique is not None assert not any("does not accept" in record.message for record in caplog.records) + + +def _make_pool() -> list[AttackTechniqueFactory]: + """A small pool of tagged factories for exercising the technique-class builder.""" + return [ + AttackTechniqueFactory(name="alpha", attack_class=_StubAttack, technique_tags=["light", "single_turn"]), + AttackTechniqueFactory(name="beta", attack_class=_StubAttack, technique_tags=["light", "multi_turn"]), + AttackTechniqueFactory(name="gamma", attack_class=_StubAttack, technique_tags=["multi_turn"]), + ] + + +class TestBuildTechniqueClassFromFactories: + """Tests for the ``build_technique_class_from_factories`` default/aggregate contract.""" + + def test_catalog_tags_auto_promote_to_aggregates(self): + """Every catalog tag in the pool becomes a selectable aggregate expanding to its techniques.""" + cls = AttackTechniqueRegistry.build_technique_class_from_factories( + class_name="AutoAggTechnique", + factories=_make_pool(), + ) + aggregates = cls.get_aggregate_tags() + assert {"all", "light", "single_turn", "multi_turn"} <= aggregates + assert {c.value for c in cls.expand({cls("light")})} == {"alpha", "beta"} + + def test_default_tags_builds_default_aggregate_and_default_returns_it(self): + """``default_tags`` builds the DEFAULT aggregate; ``default()`` returns it, expanding to tagged techniques.""" + cls = AttackTechniqueRegistry.build_technique_class_from_factories( + class_name="DefaultTagsTechnique", + factories=_make_pool(), + default_tags={"light"}, + ) + assert cls.default() == cls("default") + assert {c.value for c in cls.expand({cls.default()})} == {"alpha", "beta"} + + def test_default_names_builds_default_aggregate_from_names(self): + """``default_names`` builds the DEFAULT aggregate from exact names.""" + cls = AttackTechniqueRegistry.build_technique_class_from_factories( + class_name="DefaultNamesTechnique", + factories=_make_pool(), + default_names={"gamma"}, + ) + assert cls.default() == cls("default") + assert {c.value for c in cls.expand({cls.default()})} == {"gamma"} + + def test_no_default_leaves_attribute_unset_and_falls_back_to_all(self): + """With no default, the builder records nothing and ``default()`` owns the single ALL fallback.""" + cls = AttackTechniqueRegistry.build_technique_class_from_factories( + class_name="NoDefaultTechnique", + factories=_make_pool(), + ) + assert not hasattr(cls, "_default_technique_value") + assert "default" not in cls.get_aggregate_tags() + assert cls.default() == cls["ALL"] + + def test_default_selection_matching_nothing_falls_back_to_all(self): + """A default set that matches no pool technique builds no DEFAULT and falls back to ALL.""" + cls = AttackTechniqueRegistry.build_technique_class_from_factories( + class_name="EmptyDefaultTechnique", + factories=_make_pool(), + default_tags={"nonexistent"}, + ) + assert not hasattr(cls, "_default_technique_value") + assert cls.default() == cls["ALL"] + + def test_both_default_tags_and_names_raises(self): + """``default_tags`` and ``default_names`` are mutually exclusive.""" + with pytest.raises(ValueError, match="at most one of default_tags or default_names"): + AttackTechniqueRegistry.build_technique_class_from_factories( + class_name="BothDefaultsTechnique", + factories=_make_pool(), + default_tags={"light"}, + default_names={"gamma"}, + ) From aed4c08d9db8bed67a84f6bebc9f9cc48dccdeb5 Mon Sep 17 00:00:00 2001 From: Richard Lundeen Date: Wed, 15 Jul 2026 20:40:41 -0700 Subject: [PATCH 09/10] FIX: Satisfy pre-commit (ty ignore syntax, docstring, ruff format) - attack_technique_registry: use the repo's ty-ignore syntax (ty:unresolved-attribute) for the dynamically-set _default_technique_value - encoding.py EncodingTechnique.default(): imperative-mood docstring + Returns section - ruff-format normalization of an assert-message wrap in the registry test Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: e0bc715e-cd3a-480c-afe8-5cac8339224f --- pyrit/registry/components/attack_technique_registry.py | 2 +- pyrit/scenario/scenarios/garak/encoding.py | 7 ++++++- tests/unit/registry/test_attack_technique_registry.py | 6 +++--- 3 files changed, 10 insertions(+), 5 deletions(-) diff --git a/pyrit/registry/components/attack_technique_registry.py b/pyrit/registry/components/attack_technique_registry.py index dbcdf29958..c185f5e33f 100644 --- a/pyrit/registry/components/attack_technique_registry.py +++ b/pyrit/registry/components/attack_technique_registry.py @@ -280,7 +280,7 @@ def _get_aggregate_tags(cls: type) -> set[str]: # When it wasn't, the attribute is left unset and ScenarioTechnique.default() owns # the single ALL fallback — so the "no default -> ALL" rule lives in one place. if default_member_names: - technique_cls._default_technique_value = "default" # type: ignore[attr-defined] + technique_cls._default_technique_value = "default" # type: ignore[ty:unresolved-attribute] return technique_cls # type: ignore[ty:invalid-return-type] diff --git a/pyrit/scenario/scenarios/garak/encoding.py b/pyrit/scenario/scenarios/garak/encoding.py index 1965883436..d9e409b671 100644 --- a/pyrit/scenario/scenarios/garak/encoding.py +++ b/pyrit/scenario/scenarios/garak/encoding.py @@ -124,7 +124,12 @@ def get_aggregate_tags(cls) -> set[str]: @classmethod def default(cls) -> "EncodingTechnique": - """The out-of-the-box run selects the curated ``DEFAULT`` aggregate, not the exhaustive ``ALL``.""" + """ + Select the curated ``DEFAULT`` aggregate for the out-of-the-box run, not the exhaustive ``ALL``. + + Returns: + EncodingTechnique: The curated ``DEFAULT`` aggregate. + """ return cls.DEFAULT diff --git a/tests/unit/registry/test_attack_technique_registry.py b/tests/unit/registry/test_attack_technique_registry.py index cb9b7a64a1..ebff9d3ef7 100644 --- a/tests/unit/registry/test_attack_technique_registry.py +++ b/tests/unit/registry/test_attack_technique_registry.py @@ -303,9 +303,9 @@ def test_factory_attack_class_set(self, factory: AttackTechniqueFactory): def test_factory_attack_class_accepts_objective_target(self, factory: AttackTechniqueFactory): """Every attack class must accept ``objective_target`` (required at create time).""" sig = inspect.signature(factory.attack_class.__init__) - assert ( - "objective_target" in sig.parameters - ), f"{factory.attack_class.__name__} is missing required 'objective_target' parameter" + assert "objective_target" in sig.parameters, ( + f"{factory.attack_class.__name__} is missing required 'objective_target' parameter" + ) def test_factory_names_are_unique(self): """No two factories should share the same name.""" From bfeaeb0fa2ad92b326af017c50b93895a7b20e92 Mon Sep 17 00:00:00 2001 From: Richard Lundeen Date: Wed, 15 Jul 2026 21:00:08 -0700 Subject: [PATCH 10/10] STY: Drop redundant quotes from WebInjectionTechnique.default() annotation (ruff UP037) Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: e0bc715e-cd3a-480c-afe8-5cac8339224f --- pyrit/scenario/scenarios/garak/web_injection.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyrit/scenario/scenarios/garak/web_injection.py b/pyrit/scenario/scenarios/garak/web_injection.py index 74c480c2ef..dc29e7117e 100644 --- a/pyrit/scenario/scenarios/garak/web_injection.py +++ b/pyrit/scenario/scenarios/garak/web_injection.py @@ -199,7 +199,7 @@ def get_aggregate_tags(cls) -> set[str]: return {"all", "default", "exfil", "xss"} @classmethod - def default(cls) -> "WebInjectionTechnique": + def default(cls) -> WebInjectionTechnique: """Return the default technique (``DEFAULT``) used when the caller selects nothing.""" return cls.DEFAULT