diff --git a/pyrit/scenario/core/matrix_atomic_attack_builder.py b/pyrit/scenario/core/matrix_atomic_attack_builder.py new file mode 100644 index 0000000000..838c87d5a2 --- /dev/null +++ b/pyrit/scenario/core/matrix_atomic_attack_builder.py @@ -0,0 +1,313 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +""" +Reusable matrix builder for scenario atomic attacks. + +``MatrixAtomicAttackBuilder`` turns a technique × dataset (× optional adversarial +target) grid into a flat list of ``AtomicAttack`` instances. It centralizes the +seed-technique compatibility filtering, ``factory.create`` wiring, ``AtomicAttack`` +construction, and baseline emission for scenarios whose attacks form such a +cross-product. + +This is one member of a *family* of construction helpers named by shape +(``Matrix...``); scenarios whose construction is composite or per-objective build +``AtomicAttack`` lists differently. There is intentionally no shared builder +interface yet — each scenario calls the builder it needs directly. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass +from typing import TYPE_CHECKING, cast + +from pyrit.executor.attack import AttackScoringConfig +from pyrit.executor.attack.single_turn.prompt_sending import PromptSendingAttack +from pyrit.models import SeedAttackGroup +from pyrit.scenario.core.atomic_attack import AtomicAttack +from pyrit.scenario.core.attack_technique import AttackTechnique + +if TYPE_CHECKING: + from collections.abc import Callable, Sequence + + from pyrit.prompt_target import PromptTarget + from pyrit.scenario.core.attack_technique_factory import AttackTechniqueFactory + from pyrit.score import Scorer + from pyrit.score.true_false.true_false_scorer import TrueFalseScorer + +logger = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class MatrixCombo: + """ + One cell of the build matrix, passed to the ``name_fn``/``display_group_fn`` callbacks. + + Attributes: + technique_name (str): The technique (strategy enum value) for this cell. + dataset_name (str): The dataset key from ``DatasetAttackConfiguration.get_attack_groups_by_dataset_async()``. + target_name (str | None): The adversarial-target registry name when an + adversarial-target axis is in play, else ``None``. + """ + + technique_name: str + dataset_name: str + target_name: str | None = None + + +def _default_atomic_attack_name(combo: MatrixCombo) -> str: + """ + Default ``atomic_attack_name`` builder; target-aware so names stay unique. + + Args: + combo (MatrixCombo): The matrix cell being named. + + Returns: + str: The atomic attack name for the cell. + """ + if combo.target_name is None: + return f"{combo.technique_name}_{combo.dataset_name}" + return f"{combo.technique_name}__{combo.target_name}_{combo.dataset_name}" + + +def _default_display_group(combo: MatrixCombo) -> str: + """ + Build the default display group: aggregate results by technique. + + Args: + combo (MatrixCombo): The matrix cell being grouped. + + Returns: + str: The display group for the cell. + """ + return combo.technique_name + + +def build_baseline_atomic_attack( + *, + objective_target: PromptTarget, + objective_scorer: Scorer, + seed_groups: list[SeedAttackGroup], + memory_labels: dict[str, str] | None = None, +) -> AtomicAttack: + """ + Build the baseline ``AtomicAttack`` that sends each objective unmodified. + + The baseline is a plain ``PromptSendingAttack`` used as a comparison point against + a scenario's strategy attacks. Pass the *same* ``seed_groups`` used to build the + strategy attacks so both populations match — re-resolving under ``max_dataset_size`` + would draw a fresh random sample and diverge from the strategy population. + + Args: + objective_target (PromptTarget): The target to attack. + objective_scorer (Scorer): The scorer used to evaluate the baseline. + seed_groups (list[SeedAttackGroup]): Seed groups to attack. Used as-is. + memory_labels (dict[str, str] | None): Labels applied to the baseline's prompts. + + Returns: + AtomicAttack: The baseline atomic attack named ``"baseline"``. + """ + attack = PromptSendingAttack( + objective_target=objective_target, + attack_scoring_config=AttackScoringConfig(objective_scorer=cast("TrueFalseScorer", objective_scorer)), + ) + return AtomicAttack( + atomic_attack_name="baseline", + attack_technique=AttackTechnique(attack=attack), + seed_groups=seed_groups, + memory_labels=memory_labels or {}, + ) + + +class MatrixAtomicAttackBuilder: + """ + Build ``AtomicAttack`` instances from a technique × dataset (× target) cross-product. + + Construct once with the shared run inputs (target, scorer, labels), then call + ``build`` with the per-run grid. The builder owns: + + - seed-technique compatibility filtering (``SeedAttackGroup.filter_compatible``), + - the ``factory.create(...)`` call, forwarding an adversarial target when the + adversarial-target axis is active, + - ``AtomicAttack`` construction with naming and display-group stamping, and + - optional baseline emission using the same resolved seed groups. + + Example: + >>> builder = MatrixAtomicAttackBuilder( + ... objective_target=target, + ... objective_scorer=scorer, + ... memory_labels=labels, + ... ) + >>> attacks = builder.build( + ... technique_factories=factories, + ... dataset_groups=groups, + ... include_baseline=True, + ... ) + """ + + def __init__( + self, + *, + objective_target: PromptTarget, + objective_scorer: Scorer, + memory_labels: dict[str, str] | None = None, + ) -> None: + """ + Initialize the builder with inputs shared across every atomic attack it produces. + + Args: + objective_target (PromptTarget): The target system to attack. + objective_scorer (Scorer): The scorer applied to each produced atomic attack + and to the baseline. + memory_labels (dict[str, str] | None): Labels applied to every produced + atomic attack. + """ + self._objective_target = objective_target + self._objective_scorer = objective_scorer + self._memory_labels = memory_labels or {} + + def build( + self, + *, + technique_factories: dict[str, AttackTechniqueFactory], + dataset_groups: dict[str, list[SeedAttackGroup]], + adversarial_targets: Sequence[tuple[str, PromptTarget]] | None = None, + name_fn: Callable[[MatrixCombo], str] | None = None, + display_group_fn: Callable[[MatrixCombo], str] | None = None, + include_baseline: bool = False, + ) -> list[AtomicAttack]: + """ + Build the atomic attacks for the given grid. + + Iterates technique → (adversarial target) → dataset. The caller pre-resolves + ``technique_factories`` to exactly the techniques to build (and, by dict + insertion order, the order to build them in), so the builder does not need the + full registry or the selected-strategy set. + + Args: + technique_factories (dict[str, AttackTechniqueFactory]): Mapping of technique + name to the factory that produces it. Only these techniques are built. + dataset_groups (dict[str, list[SeedAttackGroup]]): Mapping of dataset name to + its seed groups (e.g. ``await DatasetAttackConfiguration.get_attack_groups_by_dataset_async()``). + adversarial_targets (Sequence[tuple[str, PromptTarget]] | None): Optional + ``(name, instance)`` pairs adding an adversarial-target axis. When set, + each technique is swept across every target and the target instance is + forwarded to ``factory.create(adversarial_chat=...)``. When ``None``, the + axis is collapsed and each factory uses its own (possibly lazy) + adversarial target. + name_fn (Callable[[MatrixCombo], str] | None): Builds each ``atomic_attack_name``. + Defaults to ``"{technique}_{dataset}"`` (or ``"{technique}__{target}_{dataset}"`` + when an adversarial-target axis is active). + display_group_fn (Callable[[MatrixCombo], str] | None): Builds each + ``display_group``. Defaults to grouping by technique name. + include_baseline (bool): When ``True``, prepend a baseline atomic attack built + from the flattened seed groups across all datasets. + + Returns: + list[AtomicAttack]: The generated atomic attacks, baseline first when requested. + """ + name_fn = name_fn or _default_atomic_attack_name + display_group_fn = display_group_fn or _default_display_group + + scoring_config = AttackScoringConfig(objective_scorer=cast("TrueFalseScorer", self._objective_scorer)) + + target_axis: Sequence[tuple[str | None, PromptTarget | None]] = ( + list(adversarial_targets) if adversarial_targets else [(None, None)] + ) + + atomic_attacks: list[AtomicAttack] = [] + for technique_name, factory in technique_factories.items(): + for target_name, target_instance in target_axis: + for dataset_name, seed_groups in dataset_groups.items(): + compatible_groups = self._filter_compatible_groups( + factory=factory, + seed_groups=seed_groups, + technique_name=technique_name, + dataset_name=dataset_name, + ) + if compatible_groups is None: + continue + + create_adversarial = {"adversarial_chat": target_instance} if target_instance is not None else {} + attack_technique = factory.create( + objective_target=self._objective_target, + attack_scoring_config=scoring_config, + **create_adversarial, + ) + + combo = MatrixCombo( + technique_name=technique_name, + dataset_name=dataset_name, + target_name=target_name, + ) + atomic_attacks.append( + AtomicAttack( + atomic_attack_name=name_fn(combo), + attack_technique=attack_technique, + seed_groups=compatible_groups, + adversarial_chat=( + target_instance if target_instance is not None else factory.adversarial_chat + ), + objective_scorer=cast("TrueFalseScorer", self._objective_scorer), + memory_labels=self._memory_labels, + display_group=display_group_fn(combo), + ) + ) + + if include_baseline: + all_seed_groups = [group for groups in dataset_groups.values() for group in groups] + atomic_attacks.insert( + 0, + build_baseline_atomic_attack( + objective_target=self._objective_target, + objective_scorer=self._objective_scorer, + seed_groups=all_seed_groups, + memory_labels=self._memory_labels, + ), + ) + + return atomic_attacks + + def _filter_compatible_groups( + self, + *, + factory: AttackTechniqueFactory, + seed_groups: list[SeedAttackGroup], + technique_name: str, + dataset_name: str, + ) -> list[SeedAttackGroup] | None: + """ + Filter seed groups to those compatible with the factory's seed technique. + + Args: + factory (AttackTechniqueFactory): The factory whose ``seed_technique`` gates + compatibility. + seed_groups (list[SeedAttackGroup]): Candidate seed groups for one dataset. + technique_name (str): Technique name, used only for log messages. + dataset_name (str): Dataset name, used only for log messages. + + Returns: + list[SeedAttackGroup] | None: The compatible groups, or ``None`` when the + ``(technique, dataset)`` pair has no compatible groups and should be skipped. + """ + if factory.seed_technique is None: + return list(seed_groups) + + compatible_groups = SeedAttackGroup.filter_compatible( + seed_groups=seed_groups, + technique=factory.seed_technique, + ) + skipped = len(seed_groups) - len(compatible_groups) + if skipped: + logger.info( + f"Skipped {skipped} seed group(s) from '{dataset_name}' for technique " + f"'{technique_name}' (prompt sequences overlap with simulated conversation)." + ) + if not compatible_groups: + logger.warning( + f"No compatible seed groups in '{dataset_name}' for technique " + f"'{technique_name}', skipping this (technique, dataset) pair." + ) + return None + return compatible_groups diff --git a/pyrit/scenario/core/scenario.py b/pyrit/scenario/core/scenario.py index 09dc897bdf..85868ea909 100644 --- a/pyrit/scenario/core/scenario.py +++ b/pyrit/scenario/core/scenario.py @@ -17,7 +17,7 @@ from collections.abc import Sequence from enum import Enum from pathlib import Path -from typing import TYPE_CHECKING, Any, ClassVar, cast +from typing import TYPE_CHECKING, Any, ClassVar try: # Built-in on Python 3.11+. Fall back to the ``exceptiongroup`` backport on 3.10 @@ -32,7 +32,6 @@ from pyrit.common.deprecation import print_deprecation_message from pyrit.common.utils import to_sha256 from pyrit.executor.attack import AttackExecutor -from pyrit.executor.attack.single_turn.prompt_sending import PromptSendingAttack from pyrit.memory import CentralMemory from pyrit.memory.memory_models import ScenarioResultEntry from pyrit.models import ( @@ -48,8 +47,12 @@ from pyrit.prompt_target.common.target_requirements import TargetRequirements from pyrit.registry import ScorerRegistry from pyrit.scenario.core.atomic_attack import AtomicAttack -from pyrit.scenario.core.attack_technique import AttackTechnique from pyrit.scenario.core.dataset_configuration import DatasetAttackConfiguration +from pyrit.scenario.core.matrix_atomic_attack_builder import ( + MatrixAtomicAttackBuilder, + build_baseline_atomic_attack, +) +from pyrit.scenario.core.scenario_context import ScenarioContext from pyrit.scenario.core.scenario_strategy import ScenarioStrategy from pyrit.scenario.core.scenario_target_defaults import get_default_scorer_target from pyrit.score import ( @@ -251,6 +254,11 @@ def __init__( self._max_concurrency: int | None = None self._max_retries: int = 0 + # Effective dataset configuration for the current run. initialize_async reassigns + # this to the caller-supplied config (or the default); defaulting it here means the + # attribute always exists for context construction. + self._dataset_config: DatasetAttackConfiguration = default_dataset_config + self._objective_scorer = objective_scorer self._objective_scorer_identifier = objective_scorer.get_identifier() @@ -816,16 +824,9 @@ def _build_baseline_atomic_attack(self, *, seed_groups: list[SeedAttackGroup]) - if self._objective_scorer is None: raise ValueError("Objective scorer is required to create baseline attack.") - from pyrit.executor.attack.core.attack_config import AttackScoringConfig - - attack = PromptSendingAttack( + return build_baseline_atomic_attack( objective_target=self._objective_target, - attack_scoring_config=AttackScoringConfig(objective_scorer=cast("TrueFalseScorer", self._objective_scorer)), - ) - - return AtomicAttack( - atomic_attack_name="baseline", - attack_technique=AttackTechnique(attack=attack), + objective_scorer=self._objective_scorer, seed_groups=seed_groups, memory_labels=self._memory_labels, ) @@ -974,22 +975,39 @@ async def _get_remaining_atomic_attacks_async(self) -> list[AtomicAttack]: return remaining_attacks - async def _get_atomic_attacks_async(self) -> list[AtomicAttack]: + async def _resolve_seed_groups_by_dataset_async(self) -> dict[str, list[SeedAttackGroup]]: """ - Build atomic attacks from the cross-product of selected techniques and datasets. + Resolve the seed groups this scenario attacks, keyed by originating dataset. - Uses ``_get_attack_technique_factories()`` to obtain factories, then - iterates over every (technique, dataset) pair to create an - ``AtomicAttack`` for each. Grouping for display is controlled by - ``_build_display_group()``. + This is the single place seed resolution happens for a run. The base ``Scenario`` + calls it once in the bridge, flattens the result into ``context.seed_groups``, and + reuses the same population for every atomic attack and the baseline — so sampling + under ``max_dataset_size`` stays consistent across all of them. - Subclasses that do **not** use the factory/registry pattern should - override this method entirely. Overrides that want baseline support - must call ``self._build_baseline_atomic_attack`` with the strategy - seeds. + Override to inject seeds from an alternate source (e.g. deprecated ``objectives``) + or to filter the resolved groups before attacks are built. Returns: - list[AtomicAttack]: The generated atomic attacks. + dict[str, list[SeedAttackGroup]]: Seed groups keyed by dataset name. + """ + return await self._dataset_config.get_attack_groups_by_dataset_async() + + def _build_scenario_context(self, *, seed_groups_by_dataset: dict[str, list[SeedAttackGroup]]) -> ScenarioContext: + """ + Snapshot the resolved runtime inputs into a ``ScenarioContext``. + + Called after ``initialize_async`` has populated the objective target, scorer, + strategies, dataset config, labels, and baseline flag. The resulting context is + handed to ``_build_atomic_attacks_async`` so scenario authors never read + half-initialized ``self._*`` state to build attacks. + + Args: + seed_groups_by_dataset (dict[str, list[SeedAttackGroup]]): Seed groups already + resolved once (see ``_resolve_seed_groups_by_dataset_async``). The flat + ``context.seed_groups`` is derived from these so both views share one sample. + + Returns: + ScenarioContext: The immutable inputs for atomic-attack construction. Raises: ValueError: If the scenario has not been initialized. @@ -999,68 +1017,92 @@ async def _get_atomic_attacks_async(self) -> list[AtomicAttack]: "Scenario not properly initialized. Call await scenario.initialize_async() before running." ) - from pyrit.executor.attack import AttackScoringConfig + seed_groups = [group for groups in seed_groups_by_dataset.values() for group in groups] + + return ScenarioContext( + objective_target=self._objective_target, + scenario_strategies=tuple(self._scenario_strategies), + dataset_config=self._dataset_config, + memory_labels=dict(self._memory_labels), + include_baseline=self._include_baseline, + seed_groups=seed_groups, + seed_groups_by_dataset=seed_groups_by_dataset, + ) + + async def _get_atomic_attacks_async(self) -> list[AtomicAttack]: + """ + Build this scenario's atomic attacks (internal entry point called by ``initialize_async``). + + Resolves the seed groups once, builds a ``ScenarioContext`` from the values resolved + in ``initialize_async``, and forwards to ``_build_atomic_attacks_async`` — the extension + point scenarios override to customize attack construction. The baseline is emitted + centrally here (when ``context.include_baseline`` is set) from ``context.seed_groups``, + so overrides never re-resolve seeds or hand-roll baseline emission. This stays a stable, + no-argument entry point for ``initialize_async`` and other internal callers. + + Returns: + list[AtomicAttack]: The generated atomic attacks. + + Raises: + ValueError: If the scenario has not been initialized. + """ + seed_groups_by_dataset = await self._resolve_seed_groups_by_dataset_async() + context = self._build_scenario_context(seed_groups_by_dataset=seed_groups_by_dataset) + atomic_attacks = await self._build_atomic_attacks_async(context=context) + + # Central baseline emission. Guarded so a scenario that still emits its own baseline + # (or an aggregate that legitimately has none) isn't given a duplicate. + if context.include_baseline and (not atomic_attacks or atomic_attacks[0].atomic_attack_name != "baseline"): + atomic_attacks.insert(0, self._build_baseline_atomic_attack(seed_groups=list(context.seed_groups))) - selected_techniques = {s.value for s in self._scenario_strategies} + return atomic_attacks - factories = self._get_attack_technique_factories() - seed_groups_by_dataset = await self._dataset_config.get_attack_groups_by_dataset_async() + async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list[AtomicAttack]: + """ + Build atomic attacks from the cross-product of selected techniques and datasets. + + This is the single extension point scenarios override to map techniques, datasets, + scorers, and any extra axes into ``AtomicAttack`` instances. The default + implementation delegates to ``MatrixAtomicAttackBuilder`` using the + ``_get_attack_technique_factories()`` and ``_build_display_group()`` hooks, producing + one ``AtomicAttack`` per (technique × dataset) pair. + + Scenarios with custom construction (composite attacks, per-objective technique + selection, converter stacks) override this method and build their attacks from + ``context.seed_groups`` (or ``context.seed_groups_by_dataset``). The base owns baseline + emission, so overrides never prepend one themselves. + + Args: + context (ScenarioContext): The resolved runtime inputs for this run. - scoring_config = AttackScoringConfig(objective_scorer=cast("TrueFalseScorer", self._objective_scorer)) + Returns: + list[AtomicAttack]: The generated atomic attacks. + """ + selected_techniques = {s.value for s in context.scenario_strategies} + all_factories = self._get_attack_technique_factories() - atomic_attacks: list[AtomicAttack] = [] + technique_factories: dict[str, AttackTechniqueFactory] = {} for technique_name in selected_techniques: - factory = factories.get(technique_name) + factory = all_factories.get(technique_name) if factory is None: logger.warning(f"No factory for technique '{technique_name}', skipping.") continue + technique_factories[technique_name] = factory - for dataset_name, seed_groups in seed_groups_by_dataset.items(): - if factory.seed_technique is not None: - compatible_groups = SeedAttackGroup.filter_compatible( - seed_groups=seed_groups, - technique=factory.seed_technique, - ) - skipped = len(seed_groups) - len(compatible_groups) - if skipped: - logger.info( - f"Skipped {skipped} seed group(s) from '{dataset_name}' for technique " - f"'{technique_name}' (prompt sequences overlap with simulated conversation)." - ) - if not compatible_groups: - logger.warning( - f"No compatible seed groups in '{dataset_name}' for technique " - f"'{technique_name}', skipping this (technique, dataset) pair." - ) - continue - else: - compatible_groups = list(seed_groups) - - attack_technique = factory.create( - objective_target=self._objective_target, - attack_scoring_config=scoring_config, - ) - display_group = self._build_display_group( - technique_name=technique_name, - seed_group_name=dataset_name, - ) - atomic_attacks.append( - AtomicAttack( - atomic_attack_name=f"{technique_name}_{dataset_name}", - attack_technique=attack_technique, - seed_groups=list(compatible_groups), - adversarial_chat=factory.adversarial_chat, - objective_scorer=cast("TrueFalseScorer", self._objective_scorer), - memory_labels=self._memory_labels, - display_group=display_group, - ) - ) - - if self._include_baseline: - all_seed_groups = [g for groups in seed_groups_by_dataset.values() for g in groups] - atomic_attacks.insert(0, self._build_baseline_atomic_attack(seed_groups=all_seed_groups)) - - return atomic_attacks + builder = MatrixAtomicAttackBuilder( + objective_target=context.objective_target, + objective_scorer=self._objective_scorer, + memory_labels=context.memory_labels, + ) + return builder.build( + technique_factories=technique_factories, + dataset_groups=context.seed_groups_by_dataset, + display_group_fn=lambda combo: self._build_display_group( + technique_name=combo.technique_name, + seed_group_name=combo.dataset_name, + ), + include_baseline=False, + ) async def run_async(self) -> ScenarioResult: """ diff --git a/pyrit/scenario/core/scenario_context.py b/pyrit/scenario/core/scenario_context.py new file mode 100644 index 0000000000..28e6e8cfde --- /dev/null +++ b/pyrit/scenario/core/scenario_context.py @@ -0,0 +1,62 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +""" +Resolved runtime inputs for building a scenario's atomic attacks. + +``ScenarioContext`` is the single bundle of values a scenario needs to construct +its ``AtomicAttack`` list. The base ``Scenario`` resolves these during +``initialize_async`` and passes them to ``_build_atomic_attacks_async``, so a +scenario builds its attacks from the context it is given. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from collections.abc import Mapping, Sequence + + from pyrit.models import SeedAttackGroup + from pyrit.prompt_target import PromptTarget + from pyrit.scenario.core.dataset_configuration import DatasetAttackConfiguration + from pyrit.scenario.core.scenario_strategy import ScenarioStrategy + + +@dataclass(frozen=True) +class ScenarioContext: + """ + Immutable snapshot of the inputs needed to build a scenario's atomic attacks. + + Constructed by ``Scenario._build_scenario_context`` from the values resolved in + ``initialize_async`` and handed to ``_build_atomic_attacks_async``. A scenario + builds its attacks from this context, independent of instance-attribute + initialization order. + + Attributes: + objective_target (PromptTarget): The target system the scenario attacks. + scenario_strategies (Sequence[ScenarioStrategy]): The resolved, concrete + strategies selected for this run (aggregates already expanded). + dataset_config (DatasetAttackConfiguration): The effective dataset configuration + (caller-supplied or the scenario's default). + memory_labels (dict[str, str]): Labels applied to every attack run. + include_baseline (bool): Whether a baseline atomic attack should be emitted + for this run, already resolved against the scenario's + ``BASELINE_ATTACK_POLICY``. + seed_groups (Sequence[SeedAttackGroup]): The scenario's seed groups, resolved + and sampled once by the base ``Scenario`` (flattened across datasets). Use + these to build attacks so every atomic attack — and the baseline — draws from + the same population. + seed_groups_by_dataset (Mapping[str, list[SeedAttackGroup]]): The same resolved + seed groups keyed by originating dataset name, for scenarios that map datasets + onto separate attacks or display groups. + """ + + objective_target: PromptTarget + scenario_strategies: Sequence[ScenarioStrategy] + dataset_config: DatasetAttackConfiguration + memory_labels: dict[str, str] = field(default_factory=dict) + include_baseline: bool = False + seed_groups: Sequence[SeedAttackGroup] = field(default_factory=tuple) + seed_groups_by_dataset: Mapping[str, list[SeedAttackGroup]] = field(default_factory=dict) diff --git a/pyrit/scenario/scenarios/adaptive/adaptive_scenario.py b/pyrit/scenario/scenarios/adaptive/adaptive_scenario.py index f343097626..39ebb4ee91 100644 --- a/pyrit/scenario/scenarios/adaptive/adaptive_scenario.py +++ b/pyrit/scenario/scenarios/adaptive/adaptive_scenario.py @@ -41,6 +41,7 @@ from pyrit.prompt_target import PromptTarget from pyrit.scenario.core.attack_technique_factory import AttackTechniqueFactory from pyrit.scenario.core.dataset_configuration import DatasetAttackConfiguration + from pyrit.scenario.core.scenario_context import ScenarioContext from pyrit.scenario.core.scenario_strategy import ScenarioStrategy from pyrit.score import TrueFalseScorer @@ -159,39 +160,36 @@ def _get_attack_technique_factories(self) -> dict[str, AttackTechniqueFactory]: registry_overrides = {} return {**catalog, **registry_overrides} - async def _get_atomic_attacks_async(self) -> list[AtomicAttack]: + async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list[AtomicAttack]: """ Build one ``AtomicAttack`` per (dataset, compatible seed group) pair. For each dataset, construct a single ``AdaptiveTechniqueDispatcher`` shared across that dataset's seed groups. For each seed group, ask the dispatcher to build its per-objective ``SequentialAttack`` and - wrap it in its own ``AtomicAttack``. All dispatchers across all + wrap it in its own ``AtomicAttack``. All dispatchers across all datasets share one ``TechniqueSelector`` instance so learning accumulates globally; selection is committed up-front during scenario initialization, before any execution starts. - When ``self._include_baseline`` is true (the default under - ``BASELINE_ATTACK_POLICY = Enabled``), a baseline ``AtomicAttack`` - named ``"baseline"`` is prepended at index 0. + The base ``Scenario`` prepends the baseline ``AtomicAttack`` (named + ``"baseline"``) at index 0 when ``context.include_baseline`` is true (the + default under ``BASELINE_ATTACK_POLICY = Enabled``). + + Args: + context (ScenarioContext): The resolved runtime inputs for this run. Returns: list[AtomicAttack]: One ``AtomicAttack`` per compatible - seed group across all datasets, with the baseline (when - enabled) prepended at index 0. + seed group across all datasets. Raises: - ValueError: If ``self._objective_target`` is not set, or if - ``_build_techniques_dict`` finds no usable techniques. + ValueError: If ``_build_techniques_dict`` finds no usable techniques. """ - if self._objective_target is None: - raise ValueError("objective_target must be set before creating attacks") + techniques = self._build_techniques_dict(objective_target=context.objective_target) - techniques = self._build_techniques_dict(objective_target=self._objective_target) - - seed_groups_by_dataset = await self._dataset_config.get_attack_groups_by_dataset_async() atomic_attacks: list[AtomicAttack] = [] - for dataset_name, seed_groups in seed_groups_by_dataset.items(): + for dataset_name, seed_groups in context.seed_groups_by_dataset.items(): atomic_attacks.extend( await self._build_atomics_for_dataset_async( dataset_name=dataset_name, @@ -201,10 +199,6 @@ async def _get_atomic_attacks_async(self) -> list[AtomicAttack]: ) ) - if self._include_baseline: - all_seed_groups = [g for groups in seed_groups_by_dataset.values() for g in groups] - atomic_attacks.insert(0, self._build_baseline_atomic_attack(seed_groups=all_seed_groups)) - return atomic_attacks def _build_techniques_dict( @@ -356,7 +350,7 @@ async def _build_atomics_for_dataset_async( Raises: ValueError: If ``self._objective_target`` is not set - (defensive guard; ``_get_atomic_attacks_async`` enforces + (defensive guard; ``_build_atomic_attacks_async`` enforces this earlier). """ if self._objective_target is None: # pragma: no cover - defensive diff --git a/pyrit/scenario/scenarios/adaptive/dispatcher.py b/pyrit/scenario/scenarios/adaptive/dispatcher.py index 7005e89d76..2469029537 100644 --- a/pyrit/scenario/scenarios/adaptive/dispatcher.py +++ b/pyrit/scenario/scenarios/adaptive/dispatcher.py @@ -93,7 +93,7 @@ class AdaptiveTechniqueDispatcher: reference across all calls in a scenario so learning accumulates across objectives — though all selections are committed up-front during scenario initialization (see - ``AdaptiveScenario._get_atomic_attacks_async``). + ``AdaptiveScenario._build_atomic_attacks_async``). """ def __init__( diff --git a/pyrit/scenario/scenarios/airt/jailbreak.py b/pyrit/scenario/scenarios/airt/jailbreak.py index 3bb3a66506..9299a5406a 100644 --- a/pyrit/scenario/scenarios/airt/jailbreak.py +++ b/pyrit/scenario/scenarios/airt/jailbreak.py @@ -26,6 +26,7 @@ DatasetAttackConfiguration, ) from pyrit.scenario.core.scenario import Scenario +from pyrit.scenario.core.scenario_context import ScenarioContext from pyrit.scenario.core.scenario_strategy import ScenarioStrategy from pyrit.scenario.core.scenario_target_defaults import get_default_adversarial_target from pyrit.score import ( @@ -171,9 +172,6 @@ def __init__( ) self._legacy_include_baseline = include_baseline - # Will be resolved in _get_atomic_attacks_async - self._seed_groups: list[SeedAttackGroup] | None = None - def _get_or_create_adversarial_target(self) -> PromptTarget: """ Return the shared adversarial target, creating it on first access. @@ -188,20 +186,8 @@ def _get_or_create_adversarial_target(self) -> PromptTarget: self._adversarial_target = get_default_adversarial_target() return self._adversarial_target - async def _resolve_seed_groups_async(self) -> list[SeedAttackGroup]: - """ - Resolve seed groups from dataset configuration. - - Returns: - list[SeedAttackGroup]: List of seed attack groups with objectives to be tested. - """ - # Use dataset_config (guaranteed to be set by initialize_async). Auto-fetch - # populates memory first; a still-empty result raises a DatasetConstraintError - # naming the offending dataset, which we let propagate. - return list(await self._dataset_config.get_seed_attack_groups_async()) - async def _get_atomic_attack_from_strategy_async( - self, *, strategy: str, jailbreak_template_name: str + self, *, strategy: str, jailbreak_template_name: str, seed_groups: list[SeedAttackGroup] ) -> AtomicAttack: """ Create an atomic attack for a specific jailbreak template. @@ -209,6 +195,7 @@ async def _get_atomic_attack_from_strategy_async( Args: strategy (str): JailbreakStrategy to use. jailbreak_template_name (str): Name of the jailbreak template file. + seed_groups (list[SeedAttackGroup]): Seed groups the attack draws from. Returns: AtomicAttack: An atomic attack using the specified jailbreak template. @@ -263,34 +250,32 @@ async def _get_atomic_attack_from_strategy_async( return AtomicAttack( atomic_attack_name=f"jailbreak_{template_name}", attack_technique=AttackTechnique(attack=attack), - seed_groups=self._seed_groups or [], + seed_groups=seed_groups, ) - async def _get_atomic_attacks_async(self) -> list[AtomicAttack]: + async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list[AtomicAttack]: """ Generate atomic attacks for each jailbreak template. This method creates an atomic attack for each retrieved jailbreak template. + Args: + context (ScenarioContext): The resolved runtime inputs for this run. + Returns: list[AtomicAttack]: List of atomic attacks to execute, one per jailbreak template. """ atomic_attacks: list[AtomicAttack] = [] - # Retrieve seed prompts based on selected strategies - self._seed_groups = await self._resolve_seed_groups_async() - - strategies = {s.value for s in self._scenario_strategies} + seed_groups = list(context.seed_groups) + strategies = {s.value for s in context.scenario_strategies} for strategy in strategies: for template_name in self._jailbreaks: for _ in range(self._num_attempts): atomic_attack = await self._get_atomic_attack_from_strategy_async( - strategy=strategy, jailbreak_template_name=template_name + strategy=strategy, jailbreak_template_name=template_name, seed_groups=seed_groups ) atomic_attacks.append(atomic_attack) - if self._include_baseline: - atomic_attacks.insert(0, self._build_baseline_atomic_attack(seed_groups=self._seed_groups or [])) - return atomic_attacks diff --git a/pyrit/scenario/scenarios/airt/psychosocial.py b/pyrit/scenario/scenarios/airt/psychosocial.py index 0c13d22ea4..678e99bb93 100644 --- a/pyrit/scenario/scenarios/airt/psychosocial.py +++ b/pyrit/scenario/scenarios/airt/psychosocial.py @@ -35,6 +35,7 @@ DatasetConstraintError, ) from pyrit.scenario.core.scenario import Scenario +from pyrit.scenario.core.scenario_context import ScenarioContext from pyrit.scenario.core.scenario_strategy import ( ScenarioStrategy, ) @@ -71,14 +72,6 @@ class SubharmConfig: scoring_rubric_path: str -@dataclass -class ResolvedSeedData: - """Helper dataclass for resolved seed data.""" - - seed_groups: list[SeedAttackGroup] - subharm: str | None - - class PsychosocialStrategy(ScenarioStrategy): """ PsychosocialHarmsStrategy defines a set of strategies for testing model behavior @@ -257,17 +250,21 @@ def __init__( ) self._legacy_include_baseline = include_baseline - # Store deprecated objectives for later resolution in _resolve_seed_groups_async + # Store deprecated objectives for later resolution in _resolve_seed_groups_by_dataset_async self._deprecated_objectives = objectives - # Will be resolved in _get_atomic_attacks_async - self._seed_groups: list[SeedAttackGroup] | None = None - async def _resolve_seed_groups_async(self) -> ResolvedSeedData: + async def _resolve_seed_groups_by_dataset_async(self) -> dict[str, list[SeedAttackGroup]]: """ Resolve seed groups from deprecated objectives or dataset configuration. + Seeds are filtered to the harm category selected by the scenario strategies (e.g. + ``imminent_crisis``); the default ``ALL`` strategy keeps the broad ``psychosocial`` + category. The base ``Scenario`` flattens the result into ``context.seed_groups`` and + reuses it for the strategy attacks and the baseline. + Returns: - ResolvedSeedData: Contains seed groups and optional subharm category. + dict[str, list[SeedAttackGroup]]: Seed groups keyed by dataset (or a synthetic + key for deprecated inline objectives). Raises: ValueError: If both objectives and dataset_config are specified. @@ -281,15 +278,14 @@ async def _resolve_seed_groups_async(self) -> ResolvedSeedData: ) if self._deprecated_objectives is not None: - return ResolvedSeedData( - seed_groups=[SeedAttackGroup(seeds=[SeedObjective(value=obj)]) for obj in self._deprecated_objectives], - subharm=None, - ) + return { + "objectives": [SeedAttackGroup(seeds=[SeedObjective(value=obj)]) for obj in self._deprecated_objectives] + } harm_category_filter = self._extract_harm_category_filter() # Auto-fetch populates memory first; a still-empty result raises a # DatasetConstraintError naming the offending dataset, which we let propagate. - seed_groups = await self._dataset_config.get_seed_attack_groups_async() + seed_groups = list(await self._dataset_config.get_seed_attack_groups_async()) if harm_category_filter: seed_groups = self._filter_by_harm_category( @@ -305,10 +301,7 @@ async def _resolve_seed_groups_async(self) -> ResolvedSeedData: f"No seeds remained after filtering by harm_category '{harm_category_filter}'." ) - return ResolvedSeedData( - seed_groups=list(seed_groups), - subharm=harm_category_filter, - ) + return {harm_category_filter or "psychosocial": seed_groups} def _extract_harm_category_filter(self) -> str | None: """ @@ -404,36 +397,30 @@ def _get_scorer(self, subharm: str | None = None) -> FloatScaleThresholdScorer: # threshold=1.0 means only raw score 5 counts as attack success (extremely harmful) return FloatScaleThresholdScorer(scorer=conversation_scorer, threshold=1.0) - async def _get_atomic_attacks_async(self) -> list[AtomicAttack]: - if self._objective_target is None: - raise ValueError("objective_target must be set before creating attacks") + async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list[AtomicAttack]: try: - CHAT_TARGET_REQUIREMENTS.validate(target=self._objective_target) + CHAT_TARGET_REQUIREMENTS.validate(target=context.objective_target) except ValueError as exc: raise TypeError( f"PsychosocialHarmsScenario requires a target that supports multi-turn " - f"conversations with editable history. Target {type(self._objective_target).__name__} " + f"conversations with editable history. Target {type(context.objective_target).__name__} " f"does not satisfy these requirements: {exc}" ) from exc - resolved = await self._resolve_seed_groups_async() - self._seed_groups = resolved.seed_groups - scoring_config = self._create_scoring_config(resolved.subharm) + # Deprecated inline objectives carry no harm category, so they map to no subharm rubric. + subharm = None if self._deprecated_objectives is not None else self._extract_harm_category_filter() + seed_groups = list(context.seed_groups) + scoring_config = self._create_scoring_config(subharm) - atomic_attacks: list[AtomicAttack] = [ - *self._create_single_turn_attacks(scoring_config=scoring_config, seed_groups=self._seed_groups), + return [ + *self._create_single_turn_attacks(scoring_config=scoring_config, seed_groups=seed_groups), self._create_multi_turn_attack( scoring_config=scoring_config, - subharm=resolved.subharm, - seed_groups=self._seed_groups, + subharm=subharm, + seed_groups=seed_groups, ), ] - if self._include_baseline: - atomic_attacks.insert(0, self._build_baseline_atomic_attack(seed_groups=self._seed_groups)) - - return atomic_attacks - def _create_scoring_config(self, subharm: str | None) -> AttackScoringConfig: subharm_config = self._subharm_configs.get(subharm) if subharm else None scorer = self._get_scorer(subharm=subharm) if subharm_config else self._objective_scorer diff --git a/pyrit/scenario/scenarios/airt/scam.py b/pyrit/scenario/scenarios/airt/scam.py index 69896813ba..6c8fa05b9d 100644 --- a/pyrit/scenario/scenarios/airt/scam.py +++ b/pyrit/scenario/scenarios/airt/scam.py @@ -29,6 +29,7 @@ DatasetAttackConfiguration, ) from pyrit.scenario.core.scenario import Scenario +from pyrit.scenario.core.scenario_context import ScenarioContext from pyrit.scenario.core.scenario_strategy import ScenarioStrategy from pyrit.scenario.core.scenario_target_defaults import get_default_adversarial_target from pyrit.score import TrueFalseScorer @@ -167,27 +168,13 @@ def __init__( ) self._legacy_include_baseline = include_baseline - # Will be resolved in _get_atomic_attacks_async - self._seed_groups: list[SeedAttackGroup] | None = None - - async def _resolve_seed_groups_async(self) -> list[SeedAttackGroup]: - """ - Resolve seed groups from dataset configuration. - - Returns: - list[SeedAttackGroup]: List of seed attack groups with objectives to be tested. - """ - # Use dataset_config (guaranteed to be set by initialize_async). Auto-fetch - # populates memory first; a still-empty result raises a DatasetConstraintError - # naming the offending dataset, which we let propagate. - return list(await self._dataset_config.get_seed_attack_groups_async()) - - def _get_atomic_attack_from_strategy(self, strategy: str) -> AtomicAttack: + def _get_atomic_attack_from_strategy(self, *, strategy: str, seed_groups: list[SeedAttackGroup]) -> AtomicAttack: """ Translate the strategies into actual AtomicAttacks. Args: strategy (str): The strategy to create the attack from. + seed_groups (list[SeedAttackGroup]): Seed groups the attack draws from. Returns: AtomicAttack: Configured for the specified strategy. @@ -236,25 +223,23 @@ def _get_atomic_attack_from_strategy(self, strategy: str) -> AtomicAttack: return AtomicAttack( atomic_attack_name=f"scam_{strategy}", attack_technique=AttackTechnique(attack=attack_strategy), - seed_groups=self._seed_groups or [], + seed_groups=seed_groups, memory_labels=self._memory_labels, ) - async def _get_atomic_attacks_async(self) -> list[AtomicAttack]: + async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list[AtomicAttack]: """ Generate atomic attacks for each strategy. + Args: + context (ScenarioContext): The resolved runtime inputs for this run. + Returns: list[AtomicAttack]: List of atomic attacks to execute. """ - # Resolve seed groups from deprecated objectives or dataset config - self._seed_groups = await self._resolve_seed_groups_async() - - strategies = {s.value for s in self._scenario_strategies} - - atomic_attacks = [self._get_atomic_attack_from_strategy(strategy) for strategy in strategies] + seed_groups = list(context.seed_groups) + strategies = {s.value for s in context.scenario_strategies} - if self._include_baseline: - atomic_attacks.insert(0, self._build_baseline_atomic_attack(seed_groups=self._seed_groups or [])) - - return atomic_attacks + return [ + self._get_atomic_attack_from_strategy(strategy=strategy, seed_groups=seed_groups) for strategy in strategies + ] diff --git a/pyrit/scenario/scenarios/benchmark/adversarial.py b/pyrit/scenario/scenarios/benchmark/adversarial.py index b7490e56f2..c13d1e5b59 100644 --- a/pyrit/scenario/scenarios/benchmark/adversarial.py +++ b/pyrit/scenario/scenarios/benchmark/adversarial.py @@ -11,23 +11,23 @@ from pyrit.analytics import get_cached_results_for_technique from pyrit.common import apply_defaults -from pyrit.executor.attack import AttackScoringConfig from pyrit.models import ( AttackOutcome, AttackResult, ObjectiveTargetEvaluationIdentifier, ScenarioResult, - SeedAttackGroup, ) from pyrit.models.parameter import Parameter from pyrit.registry import AttackTechniqueRegistry, TargetRegistry from pyrit.registry.tag_query import TagQuery -from pyrit.scenario.core.atomic_attack import AtomicAttack from pyrit.scenario.core.dataset_configuration import DatasetAttackConfiguration +from pyrit.scenario.core.matrix_atomic_attack_builder import MatrixAtomicAttackBuilder from pyrit.scenario.core.scenario import BaselineAttackPolicy, Scenario if TYPE_CHECKING: from pyrit.prompt_target import PromptTarget + from pyrit.scenario.core.atomic_attack import AtomicAttack + from pyrit.scenario.core.scenario_context import ScenarioContext from pyrit.scenario.core.scenario_strategy import ScenarioStrategy from pyrit.score.true_false.true_false_scorer import TrueFalseScorer @@ -51,7 +51,7 @@ def _build_benchmark_strategy() -> type[ScenarioStrategy]: / ``multi_turn`` aggregates derived from each factory's ``strategy_tags``. The (technique × target) cross-product is materialized lazily in - ``AdversarialBenchmark._get_atomic_attacks_async`` from the + ``AdversarialBenchmark._build_atomic_attacks_async`` from the user-supplied ``adversarial_targets`` parameter. Returns: @@ -85,7 +85,7 @@ class AdversarialBenchmark(Scenario): ``TargetInitializer`` from ``ADVERSARIAL_CHAT_*`` env vars, or programmatically via ``TargetRegistry.get_registry_singleton().instances.register``. - At run time, ``_get_atomic_attacks_async`` performs the + At run time, ``_build_atomic_attacks_async`` performs the ``(technique × adversarial_target × dataset)`` cross-product: for each selected adversarial-capable ``core`` factory in the ``AttackTechniqueRegistry`` and each requested target, it calls @@ -114,7 +114,7 @@ def supported_parameters(cls) -> list[Parameter]: Declare the ``adversarial_targets`` parameter. The list is treated as required at run time: - ``_get_atomic_attacks_async`` raises ``ValueError`` if + ``_build_atomic_attacks_async`` raises ``ValueError`` if ``self.params["adversarial_targets"]`` is empty or missing. The scenario-side error (rather than a declaration-side default) lets the caller raise a domain-specific message that names the CLI flag, @@ -158,7 +158,7 @@ def __init__( up by an initializer). Widening to general ``Scorer`` support (covering ``FloatScaleScorer``, etc.) is tracked as a follow-up. - use_cached: When ``True``, ``_get_atomic_attacks_async`` filters + use_cached: When ``True``, ``_build_atomic_attacks_async`` filters out atomic attacks for which the live behavioral cache (``pyrit.analytics.get_cached_results_for_technique``) has already returned at least one ``SUCCESS`` or ``FAILURE`` @@ -194,36 +194,31 @@ def __init__( scenario_result_id=scenario_result_id, ) - async def _get_atomic_attacks_async(self) -> list[AtomicAttack]: + async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list[AtomicAttack]: """ Build atomic attacks from (technique × adversarial_target × dataset), then apply caching. - Reads the user-supplied ``adversarial_targets`` parameter, resolves - each name to a ``PromptTarget`` via ``TargetRegistry``, and - cross-products the selected adversarial-capable techniques over the - resolved targets and configured datasets. Each pair calls - ``factory.create(adversarial_chat=...)`` with the - resolved target — no global registry state is touched. When - ``self._use_cached`` is set, the final candidate list is filtered - against the live behavioral cache via + Reads the user-supplied ``adversarial_targets`` parameter, resolves each name to a + ``PromptTarget`` via ``TargetRegistry``, and delegates the + ``(technique × target × dataset)`` cross-product to ``MatrixAtomicAttackBuilder`` + with the resolved targets as its adversarial-target axis. Each pair calls + ``factory.create(adversarial_chat=...)`` with the resolved target — no global + registry state is touched. When ``self._use_cached`` is set, the resulting candidate + list is filtered against the live behavioral cache via ``_collect_cached_completion_pairs``, which delegates to - ``pyrit.analytics.get_cached_results_for_technique`` for each - unique ``(technique_eval_hash, objective_target_eval_hash)`` pair. + ``pyrit.analytics.get_cached_results_for_technique`` for each unique + ``(technique_eval_hash, objective_target_eval_hash)`` pair. + + Args: + context (ScenarioContext): The resolved runtime inputs for this run. Returns: - list[AtomicAttack]: The atomic attacks to actually execute on - this run. + list[AtomicAttack]: The atomic attacks to actually execute on this run. Raises: - ValueError: If the scenario has not been initialized, if - ``adversarial_targets`` is missing/empty, or if any name in + ValueError: If ``adversarial_targets`` is missing/empty, or if any name in ``adversarial_targets`` is not registered. """ - if self._objective_target is None: - raise ValueError( - "Scenario not properly initialized. Call await scenario.initialize_async() before running." - ) - target_names = self.params.get("adversarial_targets") if not target_names: raise ValueError( @@ -235,57 +230,28 @@ async def _get_atomic_attacks_async(self) -> list[AtomicAttack]: resolved_targets = self._resolve_adversarial_targets(target_names=target_names) all_factories = AttackTechniqueRegistry.get_registry_singleton().get_factories_or_raise() - selected_factories = [all_factories[s.value] for s in self._scenario_strategies if s.value in all_factories] - - scoring_config = AttackScoringConfig(objective_scorer=self._objective_scorer) - seed_groups_by_dataset = await self._dataset_config.get_attack_groups_by_dataset_async() - - atomic_attacks: list[AtomicAttack] = [] - for factory in selected_factories: - for target_name, target_instance in resolved_targets: - for dataset_name, seed_groups in seed_groups_by_dataset.items(): - if factory.seed_technique is not None: - compatible_groups = SeedAttackGroup.filter_compatible( - seed_groups=seed_groups, - technique=factory.seed_technique, - ) - skipped = len(seed_groups) - len(compatible_groups) - if skipped: - logger.info( - f"Skipped {skipped} seed group(s) from '{dataset_name}' for technique " - f"'{factory.name}' (prompt sequences overlap with simulated conversation)." - ) - if not compatible_groups: - logger.warning( - f"No compatible seed groups in '{dataset_name}' for technique " - f"'{factory.name}', skipping this (technique, target, dataset) triple." - ) - continue - else: - compatible_groups = list(seed_groups) - - attack_technique = factory.create( - objective_target=self._objective_target, - attack_scoring_config=scoring_config, - adversarial_chat=target_instance, - ) - # ``display_group`` is set explicitly here so result roll-ups group by the - # TargetRegistry name the caller passed via ``--adversarial-targets`` — - # not by any internal field on the PromptTarget instance (e.g. ``_model_name``). - # Because we override ``_get_atomic_attacks_async`` entirely, the base - # ``Scenario._build_display_group`` hook is never consulted; ``Scenario._finalize`` - # then reads ``aa.display_group`` directly (scenario.py:721). - atomic_attacks.append( - AtomicAttack( - atomic_attack_name=f"{factory.name}__{target_name}_{dataset_name}", - attack_technique=attack_technique, - seed_groups=list(compatible_groups), - adversarial_chat=target_instance, - objective_scorer=self._objective_scorer, - memory_labels=self._memory_labels, - display_group=target_name, - ) - ) + technique_factories = { + strategy.value: all_factories[strategy.value] + for strategy in context.scenario_strategies + if strategy.value in all_factories + } + + builder = MatrixAtomicAttackBuilder( + objective_target=context.objective_target, + objective_scorer=self._objective_scorer, + memory_labels=context.memory_labels, + ) + # ``display_group`` is the TargetRegistry name the caller passed via + # ``--adversarial-targets`` so per-model ASR rolls up naturally — not any internal + # field on the PromptTarget instance (e.g. ``_model_name``). The builder's default + # ``{technique}__{target}_{dataset}`` naming preserves the VERSION=2 cache key shape. + atomic_attacks = builder.build( + technique_factories=technique_factories, + dataset_groups=context.seed_groups_by_dataset, + adversarial_targets=resolved_targets, + display_group_fn=lambda combo: combo.target_name or "", + include_baseline=False, + ) if not self._use_cached: return atomic_attacks @@ -353,7 +319,7 @@ async def run_async(self) -> ScenarioResult: Run the scenario and merge any precomputed cached results into the returned ``ScenarioResult``. When ``use_cached=True`` skipped atomic attacks whose prior results were - loaded during ``_get_atomic_attacks_async``, this override attaches + loaded during ``_build_atomic_attacks_async``, this override attaches those results (and their display-group labels) to the live scenario result so the final report reflects both newly-executed and cache-served runs. @@ -401,12 +367,12 @@ def _collect_cached_completion_pairs(self, *, atomic_attacks: list[AtomicAttack] As a side effect, populates ``self._cached_results_by_name`` with the attribution-filtered ``AttackResult`` lists keyed by ``atomic_attack_name`` so that - ``_get_atomic_attacks_async`` can inject them into the final ``ScenarioResult`` + ``_build_atomic_attacks_async`` can inject them into the final ``ScenarioResult`` via ``run_async`` without re-filtering. Args: atomic_attacks: The candidate atomic attacks built earlier in - ``_get_atomic_attacks_async``. + ``_build_atomic_attacks_async``. Returns: set[str]: ``atomic_attack_name`` values that have at least one qualifying cached diff --git a/pyrit/scenario/scenarios/foundry/red_team_agent.py b/pyrit/scenario/scenarios/foundry/red_team_agent.py index 4c364f7307..ab9384202b 100644 --- a/pyrit/scenario/scenarios/foundry/red_team_agent.py +++ b/pyrit/scenario/scenarios/foundry/red_team_agent.py @@ -64,6 +64,7 @@ from pyrit.scenario.core.attack_technique import AttackTechnique from pyrit.scenario.core.dataset_configuration import DatasetAttackConfiguration from pyrit.scenario.core.scenario import Scenario +from pyrit.scenario.core.scenario_context import ScenarioContext from pyrit.scenario.core.scenario_strategy import ScenarioCompositeStrategy, ScenarioStrategy from pyrit.scenario.core.scenario_target_defaults import get_default_adversarial_target @@ -389,39 +390,32 @@ def _strategy_to_composite(strategy: ScenarioStrategy) -> "FoundryComposite": return FoundryComposite(attack=strategy) return FoundryComposite(attack=None, converters=[strategy]) - async def _resolve_seed_groups_async(self) -> list[SeedAttackGroup]: + async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list[AtomicAttack]: """ - Resolve seed groups from the dataset configuration. + Build one ``AtomicAttack`` per resolved FoundryComposite. - Returns: - list[SeedGroup]: The resolved seed groups. - """ - return await self._dataset_config.get_seed_attack_groups_async() - - async def _get_atomic_attacks_async(self) -> list[AtomicAttack]: - """ - Retrieve the list of AtomicAttack instances in this scenario. + Args: + context (ScenarioContext): The resolved runtime inputs for this run. Returns: list[AtomicAttack]: The list of AtomicAttack instances in this scenario. """ - # Resolve seed groups now that initialize_async has been called - self._seed_groups = await self._resolve_seed_groups_async() - - atomic_attacks = [self._get_attack_from_strategy(composition) for composition in self._scenario_composites] - - if self._include_baseline: - atomic_attacks.insert(0, self._build_baseline_atomic_attack(seed_groups=self._seed_groups)) - - return atomic_attacks - - def _get_attack_from_strategy(self, composite: FoundryComposite) -> AtomicAttack: + seed_groups = list(context.seed_groups) + return [ + self._get_attack_from_strategy(composite=composition, seed_groups=seed_groups) + for composition in self._scenario_composites + ] + + def _get_attack_from_strategy( + self, *, composite: FoundryComposite, seed_groups: list[SeedAttackGroup] + ) -> AtomicAttack: """ Get an atomic attack for the specified FoundryComposite. Args: composite (FoundryComposite): Typed composite with an optional attack strategy and zero or more converter strategies. + seed_groups (list[SeedAttackGroup]): Seed groups the attack draws from. Returns: AtomicAttack: The configured atomic attack. @@ -497,7 +491,7 @@ def _get_attack_from_strategy(self, composite: FoundryComposite) -> AtomicAttack return AtomicAttack( atomic_attack_name=composite.name, attack_technique=AttackTechnique(attack=attack), - seed_groups=self._seed_groups, + seed_groups=seed_groups, adversarial_chat=self._adversarial_chat, objective_scorer=self._attack_scoring_config.objective_scorer, memory_labels=self._memory_labels, diff --git a/pyrit/scenario/scenarios/garak/encoding.py b/pyrit/scenario/scenarios/garak/encoding.py index 57041aac1e..2cf255753a 100644 --- a/pyrit/scenario/scenarios/garak/encoding.py +++ b/pyrit/scenario/scenarios/garak/encoding.py @@ -39,6 +39,7 @@ DatasetAttackConfiguration, ) from pyrit.scenario.core.scenario import Scenario +from pyrit.scenario.core.scenario_context import ScenarioContext from pyrit.scenario.core.scenario_strategy import ScenarioStrategy from pyrit.score import TrueFalseScorer from pyrit.score.true_false.decoding_scorer import DecodingScorer @@ -186,48 +187,33 @@ def __init__( ) self._legacy_include_baseline = include_baseline - # Will be resolved in _get_atomic_attacks_async - self._resolved_seed_groups: list[SeedAttackGroup] | None = None - - async def _resolve_seed_groups_async(self) -> list[SeedAttackGroup]: + async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list[AtomicAttack]: """ - Resolve seed groups from dataset configuration. + Build the encoding atomic attacks for this run. - Returns: - list[SeedAttackGroup]: List of seed attack groups to be encoded and tested. - """ - # Use dataset_config (guaranteed to be set by initialize_async). The configured - # EncodingDatasetConfiguration shapes raw seeds into objective-bearing attack - # groups via its _build_attack_groups override; auto-fetch populates memory first - # when the configured datasets aren't present. A still-empty result raises a - # DatasetConstraintError naming the offending dataset, which we let propagate. - return await self._dataset_config.get_seed_attack_groups_async() - - async def _get_atomic_attacks_async(self) -> list[AtomicAttack]: - """ - Retrieve the list of AtomicAttack instances in this scenario. + Encoding builds attacks directly (one ``AtomicAttack`` per selected encoding scheme, + each fanned out over the decode templates) rather than via the matrix builder, since + its axis is converter configurations, not techniques. + + Args: + context (ScenarioContext): The resolved runtime inputs for this run. Returns: list[AtomicAttack]: The list of AtomicAttack instances in this scenario. """ - # Resolve seed prompts from deprecated parameter or dataset config - self._resolved_seed_groups = await self._resolve_seed_groups_async() - - atomic_attacks = self._get_converter_attacks() - - if self._include_baseline: - atomic_attacks.insert(0, self._build_baseline_atomic_attack(seed_groups=self._resolved_seed_groups or [])) - - return atomic_attacks + return self._get_converter_attacks(seed_groups=list(context.seed_groups)) # These are the same as Garak encoding attacks - def _get_converter_attacks(self) -> list[AtomicAttack]: + def _get_converter_attacks(self, *, seed_groups: list[SeedAttackGroup]) -> list[AtomicAttack]: """ Get all converter-based atomic attacks. Creates atomic attacks for each encoding scheme specified in the scenario strategies. Each encoding scheme is tested both with and without explicit decoding instructions. + Args: + seed_groups (list[SeedAttackGroup]): Seed groups the attacks draw from. + Returns: list[AtomicAttack]: List of all atomic attacks to execute. """ @@ -264,10 +250,14 @@ def _get_converter_attacks(self) -> list[AtomicAttack]: atomic_attacks = [] for conv, name in converters_with_encodings: - atomic_attacks.extend(self._get_prompt_attacks(converters=conv, encoding_name=name)) + atomic_attacks.extend( + self._get_prompt_attacks(converters=conv, encoding_name=name, seed_groups=seed_groups) + ) return atomic_attacks - def _get_prompt_attacks(self, *, converters: list[PromptConverter], encoding_name: str) -> list[AtomicAttack]: + def _get_prompt_attacks( + self, *, converters: list[PromptConverter], encoding_name: str, seed_groups: list[SeedAttackGroup] + ) -> list[AtomicAttack]: """ Create atomic attacks for a specific encoding scheme. @@ -280,6 +270,7 @@ def _get_prompt_attacks(self, *, converters: list[PromptConverter], encoding_nam Args: converters (list[PromptConverter]): The list of converters to apply to the seed prompts. encoding_name (str): Human-readable name of the encoding scheme (e.g., "Base64", "ROT13"). + seed_groups (list[SeedAttackGroup]): Seed groups the attacks draw from. Returns: list[AtomicAttack]: List of atomic attacks for this encoding scheme. @@ -318,7 +309,7 @@ def _get_prompt_attacks(self, *, converters: list[PromptConverter], encoding_nam AtomicAttack( atomic_attack_name=encoding_name, attack_technique=AttackTechnique(attack=attack), - seed_groups=self._resolved_seed_groups or [], + seed_groups=seed_groups, ) ) diff --git a/tests/unit/scenario/airt/test_jailbreak.py b/tests/unit/scenario/airt/test_jailbreak.py index 888967d411..6376641988 100644 --- a/tests/unit/scenario/airt/test_jailbreak.py +++ b/tests/unit/scenario/airt/test_jailbreak.py @@ -147,7 +147,10 @@ class TestJailbreakInitialization: def test_init_with_scenario_result_id(self, mock_scenario_result_id): """Test initialization with a scenario result ID.""" with patch.object( - Jailbreak, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + Jailbreak, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = Jailbreak(scenario_result_id=mock_scenario_result_id) assert scenario._scenario_result_id == mock_scenario_result_id @@ -155,7 +158,10 @@ def test_init_with_scenario_result_id(self, mock_scenario_result_id): def test_init_with_default_scorer(self, mock_memory_seed_groups): """Test initialization with default scorer.""" with patch.object( - Jailbreak, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + Jailbreak, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = Jailbreak() assert scenario._objective_scorer_identifier @@ -163,7 +169,10 @@ def test_init_with_default_scorer(self, mock_memory_seed_groups): def test_init_with_custom_scorer(self, mock_objective_scorer, mock_memory_seed_groups): """Test initialization with custom scorer.""" with patch.object( - Jailbreak, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + Jailbreak, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = Jailbreak(objective_scorer=mock_objective_scorer) assert scenario._objective_scorer == mock_objective_scorer @@ -171,7 +180,10 @@ def test_init_with_custom_scorer(self, mock_objective_scorer, mock_memory_seed_g def test_init_with_num_templates(self, mock_random_num_templates): """Test initialization with num_templates provided.""" with patch.object( - Jailbreak, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + Jailbreak, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = Jailbreak(num_templates=mock_random_num_templates) assert scenario._num_templates == mock_random_num_templates @@ -179,7 +191,10 @@ def test_init_with_num_templates(self, mock_random_num_templates): def test_init_with_num_attempts(self, mock_random_num_attempts): """Test initialization with n provided.""" with patch.object( - Jailbreak, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + Jailbreak, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = Jailbreak(num_attempts=mock_random_num_attempts) assert scenario._num_attempts == mock_random_num_attempts @@ -200,7 +215,10 @@ def test_init_accepts_subdirectory_jailbreak_names(self, mock_objective_scorer, subdir_name = subdir_templates[0] with patch.object( - Jailbreak, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + Jailbreak, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = Jailbreak(objective_scorer=mock_objective_scorer, jailbreak_names=[subdir_name]) assert scenario._jailbreaks == [subdir_name] @@ -230,7 +248,10 @@ async def test_default_initialize_includes_baseline( ): """initialize_async without include_baseline honors BASELINE_ATTACK_POLICY=Enabled.""" with patch.object( - Jailbreak, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + Jailbreak, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = Jailbreak(objective_scorer=mock_objective_scorer) await scenario.initialize_async(objective_target=mock_objective_target) @@ -241,7 +262,10 @@ async def test_explicit_include_baseline_false_omits_baseline( ): """Caller can opt out of baseline by passing include_baseline=False.""" with patch.object( - Jailbreak, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + Jailbreak, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = Jailbreak(objective_scorer=mock_objective_scorer) await scenario.initialize_async( @@ -260,7 +284,10 @@ async def test_attack_generation_for_simple( ): """Test that the simple attack generation works.""" with patch.object( - Jailbreak, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + Jailbreak, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = Jailbreak(objective_scorer=mock_objective_scorer, num_templates=2) @@ -276,7 +303,10 @@ async def test_attack_generation_for_complex( ): """Test that the complex attack generation works.""" with patch.object( - Jailbreak, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + Jailbreak, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = Jailbreak(objective_scorer=mock_objective_scorer, num_templates=2) @@ -296,7 +326,10 @@ async def test_attack_generation_for_manyshot( ): """Test that the manyshot attack generation works.""" with patch.object( - Jailbreak, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + Jailbreak, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = Jailbreak(objective_scorer=mock_objective_scorer, num_templates=2) @@ -314,7 +347,10 @@ async def test_attack_generation_for_promptsending( ): """Test that the prompt sending attack generation works.""" with patch.object( - Jailbreak, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + Jailbreak, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = Jailbreak(objective_scorer=mock_objective_scorer, num_templates=2) @@ -332,7 +368,10 @@ async def test_attack_generation_for_skeleton( ): """Test that the skelton key attack generation works.""" with patch.object( - Jailbreak, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + Jailbreak, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = Jailbreak(objective_scorer=mock_objective_scorer, num_templates=2) @@ -350,7 +389,10 @@ async def test_attack_generation_for_roleplay( ): """Test that the roleplaying attack generation works.""" with patch.object( - Jailbreak, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + Jailbreak, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = Jailbreak(objective_scorer=mock_objective_scorer, num_templates=2) @@ -372,7 +414,10 @@ async def test_attack_runs_include_objectives( Combined coverage previously split across test_get_atomic_attacks_async_returns_attacks. """ with patch.object( - Jailbreak, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + Jailbreak, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = Jailbreak(objective_scorer=mock_objective_scorer, num_templates=2) @@ -389,7 +434,10 @@ async def test_get_all_jailbreak_templates( ): """Test that all jailbreak templates are found.""" with patch.object( - Jailbreak, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + Jailbreak, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = Jailbreak( objective_scorer=mock_objective_scorer, @@ -402,7 +450,10 @@ async def test_get_some_jailbreak_templates( ): """Test that random jailbreak template selection works.""" with patch.object( - Jailbreak, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + Jailbreak, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = Jailbreak(objective_scorer=mock_objective_scorer, num_templates=mock_random_num_templates) await scenario.initialize_async(objective_target=mock_objective_target) @@ -413,7 +464,10 @@ async def test_custom_num_attempts( ): """Test that n successfully tries each jailbreak template n-many times.""" with patch.object( - Jailbreak, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + Jailbreak, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): base_scenario = Jailbreak(objective_scorer=mock_objective_scorer, num_templates=2) await base_scenario.initialize_async(objective_target=mock_objective_target, include_baseline=False) @@ -443,7 +497,10 @@ async def test_initialize_async_with_max_concurrency( ) -> None: """Test initialization with custom max_concurrency.""" with patch.object( - Jailbreak, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + Jailbreak, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = Jailbreak(objective_scorer=mock_objective_scorer) await scenario.initialize_async(objective_target=mock_objective_target, max_concurrency=20) @@ -460,7 +517,10 @@ async def test_initialize_async_with_memory_labels( memory_labels = {"type": "jailbreak", "category": "scenario"} with patch.object( - Jailbreak, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + Jailbreak, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = Jailbreak(objective_scorer=mock_objective_scorer) await scenario.initialize_async( @@ -496,7 +556,10 @@ async def test_no_target_duplication_async( ) -> None: """Test that all three targets (adversarial, object, scorer) are distinct.""" with patch.object( - Jailbreak, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + Jailbreak, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = Jailbreak() await scenario.initialize_async(objective_target=mock_objective_target) @@ -543,7 +606,10 @@ async def test_roleplay_attacks_share_adversarial_target( ) -> None: """Test that multiple role-play attacks share the same adversarial target instance.""" with patch.object( - Jailbreak, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + Jailbreak, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = Jailbreak(objective_scorer=mock_objective_scorer, num_templates=2) await scenario.initialize_async( @@ -572,8 +638,8 @@ async def test_one_resolution_call_baseline_matches_strategies( seed_groups = [SeedAttackGroup(seeds=[SeedObjective(value=f"obj{i}")]) for i in range(10)] config = DatasetAttackConfiguration(seed_groups=seed_groups, max_dataset_size=3) - first_sample = seed_groups[:3] - second_sample = seed_groups[5:8] + first_sample = [("inline", group) for group in seed_groups[:3]] + second_sample = [("inline", group) for group in seed_groups[5:8]] scenario = Jailbreak(objective_scorer=mock_objective_scorer, num_templates=1) with patch( "pyrit.scenario.core.dataset_configuration.random.sample", diff --git a/tests/unit/scenario/airt/test_psychosocial.py b/tests/unit/scenario/airt/test_psychosocial.py index 45cb6ec517..e62fbeba4b 100644 --- a/tests/unit/scenario/airt/test_psychosocial.py +++ b/tests/unit/scenario/airt/test_psychosocial.py @@ -14,7 +14,7 @@ Psychosocial, PsychosocialStrategy, ) -from pyrit.scenario.scenarios.airt.psychosocial import ResolvedSeedData, SubharmConfig +from pyrit.scenario.scenarios.airt.psychosocial import SubharmConfig from pyrit.score import FloatScaleThresholdScorer SEED_DATASETS_PATH = DATASETS_PATH / "seed_datasets" / "local" / "airt" @@ -28,9 +28,9 @@ def mock_memory_seed_groups() -> list[SeedGroup]: @pytest.fixture -def mock_resolved_seed_data(mock_memory_seed_groups) -> ResolvedSeedData: - """Create mock ResolvedSeedData for patching _resolve_seed_groups.""" - return ResolvedSeedData(seed_groups=mock_memory_seed_groups, subharm=None) +def mock_seed_groups_by_dataset(mock_memory_seed_groups) -> dict[str, list[SeedAttackGroup]]: + """Create mock by-dataset seed groups for patching _resolve_seed_groups_by_dataset_async.""" + return {"psychosocial": mock_memory_seed_groups} @pytest.fixture @@ -183,12 +183,15 @@ async def test_attack_generation_for_all( self, mock_objective_target, mock_objective_scorer, - mock_resolved_seed_data, + mock_seed_groups_by_dataset, mock_dataset_config, ): """Test that _get_atomic_attacks_async returns atomic attacks.""" with patch.object( - Psychosocial, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_resolved_seed_data + Psychosocial, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value=mock_seed_groups_by_dataset, ): scenario = Psychosocial(objective_scorer=mock_objective_scorer) @@ -203,12 +206,15 @@ async def test_attack_runs_include_objectives_async( *, mock_objective_target: PromptTarget, mock_objective_scorer: FloatScaleThresholdScorer, - mock_resolved_seed_data, + mock_seed_groups_by_dataset, mock_dataset_config, ) -> None: """Test that attack runs include objectives for each seed prompt.""" with patch.object( - Psychosocial, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_resolved_seed_data + Psychosocial, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value=mock_seed_groups_by_dataset, ): scenario = Psychosocial( objective_scorer=mock_objective_scorer, @@ -225,12 +231,15 @@ async def test_get_atomic_attacks_async_returns_attacks( *, mock_objective_target: PromptTarget, mock_objective_scorer: FloatScaleThresholdScorer, - mock_resolved_seed_data, + mock_seed_groups_by_dataset, mock_dataset_config, ) -> None: """Test that _get_atomic_attacks_async returns atomic attacks.""" with patch.object( - Psychosocial, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_resolved_seed_data + Psychosocial, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value=mock_seed_groups_by_dataset, ): scenario = Psychosocial( objective_scorer=mock_objective_scorer, @@ -251,12 +260,15 @@ async def test_initialize_async_with_max_concurrency( *, mock_objective_target: PromptTarget, mock_objective_scorer: FloatScaleThresholdScorer, - mock_resolved_seed_data, + mock_seed_groups_by_dataset, mock_dataset_config, ) -> None: """Test initialization with custom max_concurrency.""" with patch.object( - Psychosocial, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_resolved_seed_data + Psychosocial, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value=mock_seed_groups_by_dataset, ): scenario = Psychosocial(objective_scorer=mock_objective_scorer) await scenario.initialize_async( @@ -269,14 +281,17 @@ async def test_initialize_async_with_memory_labels( *, mock_objective_target: PromptTarget, mock_objective_scorer: FloatScaleThresholdScorer, - mock_resolved_seed_data, + mock_seed_groups_by_dataset, mock_dataset_config, ) -> None: """Test initialization with memory labels.""" memory_labels = {"type": "psychosocial", "category": "crisis"} with patch.object( - Psychosocial, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_resolved_seed_data + Psychosocial, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value=mock_seed_groups_by_dataset, ): scenario = Psychosocial(objective_scorer=mock_objective_scorer) await scenario.initialize_async( @@ -317,12 +332,15 @@ async def test_no_target_duplication_async( self, *, mock_objective_target: PromptTarget, - mock_resolved_seed_data, + mock_seed_groups_by_dataset, mock_dataset_config, ) -> None: """Test that all three targets (adversarial, objective, scorer) are distinct.""" with patch.object( - Psychosocial, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_resolved_seed_data + Psychosocial, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value=mock_seed_groups_by_dataset, ): scenario = Psychosocial() await scenario.initialize_async(objective_target=mock_objective_target, dataset_config=mock_dataset_config) @@ -350,12 +368,15 @@ async def test_initialize_async_invokes_target_requirements_validate( self, mock_objective_target, mock_objective_scorer, - mock_resolved_seed_data, + mock_seed_groups_by_dataset, mock_dataset_config, ): """initialize_async must delegate capability validation to TARGET_REQUIREMENTS.validate.""" with patch.object( - Psychosocial, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_resolved_seed_data + Psychosocial, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value=mock_seed_groups_by_dataset, ): scenario = Psychosocial(objective_scorer=mock_objective_scorer) with patch("pyrit.prompt_target.common.target_requirements.TargetRequirements.validate") as mock_validate: @@ -373,7 +394,7 @@ async def test_initialize_async_invokes_target_requirements_validate( async def test_initialize_async_rejects_target_missing_editable_history( self, mock_objective_scorer, - mock_resolved_seed_data, + mock_seed_groups_by_dataset, mock_dataset_config, ): """A target that does not natively support EDITABLE_HISTORY must be rejected.""" @@ -390,7 +411,10 @@ async def test_initialize_async_rejects_target_missing_editable_history( ) with patch.object( - Psychosocial, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_resolved_seed_data + Psychosocial, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value=mock_seed_groups_by_dataset, ): scenario = Psychosocial(objective_scorer=mock_objective_scorer) with pytest.raises(ValueError, match="editable_history"): diff --git a/tests/unit/scenario/airt/test_scam.py b/tests/unit/scenario/airt/test_scam.py index 0218ef8973..05b4760e8c 100644 --- a/tests/unit/scenario/airt/test_scam.py +++ b/tests/unit/scenario/airt/test_scam.py @@ -59,6 +59,7 @@ def mock_dataset_config(mock_memory_seed_groups): seed_attack_groups = list(mock_memory_seed_groups) mock_config = MagicMock(spec=DatasetAttackConfiguration) mock_config.get_seed_attack_groups_async = AsyncMock(return_value=seed_attack_groups) + mock_config.get_attack_groups_by_dataset_async = AsyncMock(return_value={"airt_scam": seed_attack_groups}) mock_config.dataset_names = ["airt_scam"] return mock_config @@ -129,7 +130,10 @@ def test_init_with_default_objectives( mock_memory_seed_groups: list[SeedAttackGroup], ) -> None: with patch.object( - Scam, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + Scam, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = Scam(objective_scorer=mock_objective_scorer) @@ -139,7 +143,10 @@ def test_init_with_default_objectives( def test_init_with_default_scorer(self, mock_memory_seed_groups) -> None: """Test initialization with default scorer.""" with patch.object( - Scam, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + Scam, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = Scam() assert scenario._objective_scorer_identifier @@ -149,7 +156,10 @@ def test_init_with_custom_scorer(self, *, mock_memory_seed_groups: list[SeedAtta scorer = MagicMock(spec=TrueFalseCompositeScorer) with patch.object( - Scam, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + Scam, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = Scam(objective_scorer=scorer) assert isinstance(scenario._scorer_config, AttackScoringConfig) @@ -158,7 +168,10 @@ def test_init_default_adversarial_chat( self, *, mock_objective_scorer: TrueFalseCompositeScorer, mock_memory_seed_groups: list[SeedAttackGroup] ) -> None: with patch.object( - Scam, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + Scam, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = Scam(objective_scorer=mock_objective_scorer) @@ -172,7 +185,10 @@ def test_init_with_adversarial_chat( adversarial_chat.get_identifier.return_value = _mock_target_id("CustomAdversary") with patch.object( - Scam, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + Scam, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = Scam( adversarial_chat=adversarial_chat, @@ -209,7 +225,10 @@ async def test_attack_generation_for_all( ): """Test that _get_atomic_attacks_async returns atomic attacks.""" with patch.object( - Scam, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + Scam, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = Scam(objective_scorer=mock_objective_scorer) @@ -363,7 +382,10 @@ async def test_initialize_async_with_max_concurrency( ) -> None: """Test initialization with custom max_concurrency.""" with patch.object( - Scam, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + Scam, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = Scam(objective_scorer=mock_objective_scorer) await scenario.initialize_async( @@ -383,7 +405,10 @@ async def test_initialize_async_with_memory_labels( memory_labels = {"type": "scam", "category": "scenario"} with patch.object( - Scam, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + Scam, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = Scam(objective_scorer=mock_objective_scorer) await scenario.initialize_async( @@ -419,7 +444,10 @@ async def test_no_target_duplication_async( ) -> None: """Test that all three targets (adversarial, object, scorer) are distinct.""" with patch.object( - Scam, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + Scam, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = Scam() await scenario.initialize_async(objective_target=mock_objective_target, dataset_config=mock_dataset_config) @@ -445,8 +473,8 @@ async def test_one_resolution_call_baseline_matches_strategies( seed_groups = [SeedAttackGroup(seeds=[SeedObjective(value=f"obj{i}")]) for i in range(10)] config = DatasetAttackConfiguration(seed_groups=seed_groups, max_dataset_size=3) - first_sample = seed_groups[:3] - second_sample = seed_groups[5:8] + first_sample = [("inline", group) for group in seed_groups[:3]] + second_sample = [("inline", group) for group in seed_groups[5:8]] with patch( "pyrit.scenario.core.dataset_configuration.random.sample", side_effect=[first_sample, second_sample], diff --git a/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py b/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py new file mode 100644 index 0000000000..8b91b37561 --- /dev/null +++ b/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py @@ -0,0 +1,283 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +"""Tests for ``MatrixAtomicAttackBuilder`` and its module-level helpers. + +The builder centralizes the technique × dataset (× optional adversarial target) +cross-product for scenarios whose attacks form such a grid. These tests pin the contract: + +* cross-product cardinality across techniques, datasets, and the optional + adversarial-target axis, +* default and custom ``atomic_attack_name`` / ``display_group`` derivation, +* ``factory.create`` adversarial-chat forwarding (and the ``AtomicAttack`` + ``adversarial_chat`` it stamps), +* seed-technique compatibility filtering (skip vs. subset), and +* optional baseline emission from the flattened seed groups. +""" + +from unittest.mock import MagicMock, patch + +import pytest + +from pyrit.models import SeedAttackGroup, SeedObjective +from pyrit.prompt_target import PromptTarget +from pyrit.scenario.core.attack_technique_factory import AttackTechniqueFactory +from pyrit.scenario.core.matrix_atomic_attack_builder import ( + MatrixAtomicAttackBuilder, + MatrixCombo, + build_baseline_atomic_attack, +) +from pyrit.score import TrueFalseScorer + + +def _mock_factory(*, name: str, seed_technique=None, adversarial_chat=None) -> MagicMock: + """Build a controllable ``AttackTechniqueFactory`` stand-in. + + ``create`` returns a fresh sentinel ``AttackTechnique`` so callers can assert + on the (objective_target, scoring_config, adversarial_chat) kwargs it received. + """ + factory = MagicMock(spec=AttackTechniqueFactory) + factory.name = name + factory.seed_technique = seed_technique + factory.adversarial_chat = adversarial_chat + factory.create.return_value = MagicMock(name=f"{name}_technique") + return factory + + +def _seed_group(*, objective: str) -> SeedAttackGroup: + return SeedAttackGroup(seeds=[SeedObjective(value=objective)]) + + +def _builder() -> MatrixAtomicAttackBuilder: + return MatrixAtomicAttackBuilder( + objective_target=MagicMock(spec=PromptTarget), + objective_scorer=MagicMock(spec=TrueFalseScorer), + memory_labels={"op": "unit"}, + ) + + +@pytest.mark.usefixtures("patch_central_database") +class TestMatrixComboNaming: + """Default ``atomic_attack_name`` / ``display_group`` helpers via ``MatrixCombo``.""" + + def test_default_name_without_target(self): + builder = _builder() + result = builder.build( + technique_factories={"tech": _mock_factory(name="tech")}, + dataset_groups={"ds": [_seed_group(objective="o1")]}, + ) + assert [a.atomic_attack_name for a in result] == ["tech_ds"] + + def test_default_name_with_target_axis(self): + builder = _builder() + result = builder.build( + technique_factories={"tech": _mock_factory(name="tech")}, + dataset_groups={"ds": [_seed_group(objective="o1")]}, + adversarial_targets=[("advA", MagicMock(spec=PromptTarget))], + ) + assert [a.atomic_attack_name for a in result] == ["tech__advA_ds"] + + def test_default_display_group_is_technique(self): + builder = _builder() + result = builder.build( + technique_factories={"tech": _mock_factory(name="tech")}, + dataset_groups={"ds": [_seed_group(objective="o1")]}, + ) + assert result[0].display_group == "tech" + + +@pytest.mark.usefixtures("patch_central_database") +class TestMatrixBuildCrossProduct: + """Cardinality and ordering of the produced cross-product.""" + + def test_two_techniques_one_dataset_no_targets(self): + builder = _builder() + result = builder.build( + technique_factories={ + "alpha": _mock_factory(name="alpha"), + "beta": _mock_factory(name="beta"), + }, + dataset_groups={"ds": [_seed_group(objective="o1")]}, + ) + assert [a.atomic_attack_name for a in result] == ["alpha_ds", "beta_ds"] + + def test_one_technique_two_targets_one_dataset(self): + builder = _builder() + result = builder.build( + technique_factories={"tech": _mock_factory(name="tech")}, + dataset_groups={"ds": [_seed_group(objective="o1")]}, + adversarial_targets=[ + ("advA", MagicMock(spec=PromptTarget)), + ("advB", MagicMock(spec=PromptTarget)), + ], + ) + assert [a.atomic_attack_name for a in result] == ["tech__advA_ds", "tech__advB_ds"] + + def test_one_technique_one_target_two_datasets(self): + builder = _builder() + result = builder.build( + technique_factories={"tech": _mock_factory(name="tech")}, + dataset_groups={ + "ds1": [_seed_group(objective="o1")], + "ds2": [_seed_group(objective="o2")], + }, + adversarial_targets=[("advA", MagicMock(spec=PromptTarget))], + ) + assert [a.atomic_attack_name for a in result] == ["tech__advA_ds1", "tech__advA_ds2"] + + +@pytest.mark.usefixtures("patch_central_database") +class TestMatrixAdversarialForwarding: + """The adversarial-target axis must drive both ``factory.create`` and ``AtomicAttack``.""" + + def test_create_called_with_adversarial_chat_per_target(self): + builder = _builder() + factory = _mock_factory(name="tech") + target_a = MagicMock(spec=PromptTarget) + target_b = MagicMock(spec=PromptTarget) + builder.build( + technique_factories={"tech": factory}, + dataset_groups={"ds": [_seed_group(objective="o1")]}, + adversarial_targets=[("advA", target_a), ("advB", target_b)], + ) + assert factory.create.call_count == 2 + injected = {call.kwargs["adversarial_chat"] for call in factory.create.call_args_list} + assert injected == {target_a, target_b} + + def test_atomic_attack_adversarial_chat_is_resolved_target(self): + builder = _builder() + target_a = MagicMock(spec=PromptTarget) + result = builder.build( + technique_factories={"tech": _mock_factory(name="tech")}, + dataset_groups={"ds": [_seed_group(objective="o1")]}, + adversarial_targets=[("advA", target_a)], + ) + assert result[0]._adversarial_chat is target_a + + def test_no_target_axis_uses_factory_adversarial_chat(self): + builder = _builder() + baked = MagicMock(spec=PromptTarget) + factory = _mock_factory(name="tech", adversarial_chat=baked) + result = builder.build( + technique_factories={"tech": factory}, + dataset_groups={"ds": [_seed_group(objective="o1")]}, + ) + # No adversarial_chat is forwarded into create() when the axis is collapsed. + assert "adversarial_chat" not in factory.create.call_args.kwargs + assert result[0]._adversarial_chat is baked + + +@pytest.mark.usefixtures("patch_central_database") +class TestMatrixCustomCallbacks: + """Custom ``name_fn`` / ``display_group_fn`` override the defaults.""" + + def test_custom_name_and_display_group(self): + builder = _builder() + result = builder.build( + technique_factories={"tech": _mock_factory(name="tech")}, + dataset_groups={"ds": [_seed_group(objective="o1")]}, + adversarial_targets=[("advA", MagicMock(spec=PromptTarget))], + name_fn=lambda combo: f"{combo.target_name}:{combo.technique_name}", + display_group_fn=lambda combo: combo.target_name or "", + ) + assert result[0].atomic_attack_name == "advA:tech" + assert result[0].display_group == "advA" + + def test_callbacks_receive_full_combo(self): + builder = _builder() + seen: list[MatrixCombo] = [] + builder.build( + technique_factories={"tech": _mock_factory(name="tech")}, + dataset_groups={"ds": [_seed_group(objective="o1")]}, + adversarial_targets=[("advA", MagicMock(spec=PromptTarget))], + name_fn=lambda combo: seen.append(combo) or "n", + ) + assert seen == [MatrixCombo(technique_name="tech", dataset_name="ds", target_name="advA")] + + +@pytest.mark.usefixtures("patch_central_database") +class TestMatrixSeedTechniqueFiltering: + """``seed_technique`` gates which seed groups (and pairs) survive.""" + + def test_incompatible_pair_is_skipped(self): + builder = _builder() + factory = _mock_factory(name="tech", seed_technique=MagicMock()) + with patch.object(SeedAttackGroup, "filter_compatible", return_value=[]): + result = builder.build( + technique_factories={"tech": factory}, + dataset_groups={"ds": [_seed_group(objective="o1")]}, + ) + assert result == [] + factory.create.assert_not_called() + + def test_partial_filter_keeps_subset(self): + builder = _builder() + factory = _mock_factory(name="tech", seed_technique=MagicMock()) + kept = _seed_group(objective="keep") + dropped = _seed_group(objective="drop") + with patch.object(SeedAttackGroup, "filter_compatible", return_value=[kept]): + result = builder.build( + technique_factories={"tech": factory}, + dataset_groups={"ds": [kept, dropped]}, + ) + assert len(result) == 1 + assert result[0]._seed_groups == [kept] + + def test_no_seed_technique_keeps_all_groups(self): + builder = _builder() + groups = [_seed_group(objective="a"), _seed_group(objective="b")] + result = builder.build( + technique_factories={"tech": _mock_factory(name="tech")}, + dataset_groups={"ds": groups}, + ) + assert result[0]._seed_groups == groups + + +@pytest.mark.usefixtures("patch_central_database") +class TestMatrixBaseline: + """Baseline emission prepends a single ``baseline`` attack over flattened seeds.""" + + def test_baseline_prepended_when_requested(self): + builder = _builder() + result = builder.build( + technique_factories={"tech": _mock_factory(name="tech")}, + dataset_groups={"ds": [_seed_group(objective="o1")]}, + include_baseline=True, + ) + assert result[0].atomic_attack_name == "baseline" + assert [a.atomic_attack_name for a in result] == ["baseline", "tech_ds"] + + def test_baseline_omitted_by_default(self): + builder = _builder() + result = builder.build( + technique_factories={"tech": _mock_factory(name="tech")}, + dataset_groups={"ds": [_seed_group(objective="o1")]}, + ) + assert all(a.atomic_attack_name != "baseline" for a in result) + + def test_baseline_flattens_all_datasets(self): + builder = _builder() + g1 = _seed_group(objective="o1") + g2 = _seed_group(objective="o2") + result = builder.build( + technique_factories={"tech": _mock_factory(name="tech")}, + dataset_groups={"ds1": [g1], "ds2": [g2]}, + include_baseline=True, + ) + assert result[0]._seed_groups == [g1, g2] + + +@pytest.mark.usefixtures("patch_central_database") +class TestBuildBaselineHelper: + """The module-level ``build_baseline_atomic_attack`` helper.""" + + def test_baseline_name_and_seed_groups(self): + groups = [_seed_group(objective="o1")] + baseline = build_baseline_atomic_attack( + objective_target=MagicMock(spec=PromptTarget), + objective_scorer=MagicMock(spec=TrueFalseScorer), + seed_groups=groups, + memory_labels={"op": "unit"}, + ) + assert baseline.atomic_attack_name == "baseline" + assert baseline._seed_groups == groups diff --git a/tests/unit/scenario/core/test_scenario_context.py b/tests/unit/scenario/core/test_scenario_context.py new file mode 100644 index 0000000000..edf234447a --- /dev/null +++ b/tests/unit/scenario/core/test_scenario_context.py @@ -0,0 +1,62 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +"""Tests for the ``ScenarioContext`` frozen dataclass. + +``ScenarioContext`` is the immutable bundle the base ``Scenario`` builds in +``initialize_async`` and hands to ``_build_atomic_attacks_async``. These tests +pin the field surface, defaults, and immutability that scenario authors rely on. +""" + +from unittest.mock import MagicMock + +import pytest + +from pyrit.prompt_target import PromptTarget +from pyrit.scenario.core.dataset_configuration import DatasetConfiguration +from pyrit.scenario.core.scenario_context import ScenarioContext + + +def _context(**overrides) -> ScenarioContext: + kwargs = { + "objective_target": MagicMock(spec=PromptTarget), + "scenario_strategies": [], + "dataset_config": MagicMock(spec=DatasetConfiguration), + } + kwargs.update(overrides) + return ScenarioContext(**kwargs) + + +def test_required_fields_are_stored(): + target = MagicMock(spec=PromptTarget) + strategies = [MagicMock()] + dataset_config = MagicMock(spec=DatasetConfiguration) + context = ScenarioContext( + objective_target=target, + scenario_strategies=strategies, + dataset_config=dataset_config, + ) + assert context.objective_target is target + assert context.scenario_strategies is strategies + assert context.dataset_config is dataset_config + + +def test_memory_labels_defaults_to_empty_dict(): + context = _context() + assert context.memory_labels == {} + + +def test_memory_labels_default_is_not_shared_between_instances(): + first = _context() + second = _context() + assert first.memory_labels is not second.memory_labels + + +def test_include_baseline_defaults_to_false(): + assert _context().include_baseline is False + + +def test_is_frozen(): + context = _context() + with pytest.raises(AttributeError): + context.objective_target = MagicMock(spec=PromptTarget) diff --git a/tests/unit/scenario/foundry/test_red_team_agent.py b/tests/unit/scenario/foundry/test_red_team_agent.py index 1799f6946b..2a97be6db3 100644 --- a/tests/unit/scenario/foundry/test_red_team_agent.py +++ b/tests/unit/scenario/foundry/test_red_team_agent.py @@ -116,7 +116,10 @@ async def test_init_with_single_strategy( ): """Test initialization with a single attack strategy.""" with patch.object( - RedTeamAgent, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + RedTeamAgent, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = RedTeamAgent( attack_scoring_config=AttackScoringConfig(objective_scorer=mock_objective_scorer), @@ -141,7 +144,10 @@ async def test_init_with_multiple_strategies( ] with patch.object( - RedTeamAgent, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + RedTeamAgent, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = RedTeamAgent( attack_scoring_config=AttackScoringConfig(objective_scorer=mock_objective_scorer), @@ -180,7 +186,10 @@ async def test_init_with_memory_labels( memory_labels = {"test": "foundry", "category": "attack"} with patch.object( - RedTeamAgent, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + RedTeamAgent, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = RedTeamAgent( attack_scoring_config=AttackScoringConfig(objective_scorer=mock_objective_scorer), @@ -205,7 +214,10 @@ def test_init_creates_default_scorer_when_not_provided( mock_get_scorer.return_value = mock_scorer_instance with patch.object( - RedTeamAgent, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + RedTeamAgent, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = RedTeamAgent() @@ -240,7 +252,10 @@ async def test_normalize_easy_strategies( ): """Test that EASY strategy expands to easy attack strategies.""" with patch.object( - RedTeamAgent, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + RedTeamAgent, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = RedTeamAgent( attack_scoring_config=AttackScoringConfig(objective_scorer=mock_objective_scorer), @@ -259,7 +274,10 @@ async def test_normalize_moderate_strategies( ): """Test that MODERATE strategy expands to moderate attack strategies.""" with patch.object( - RedTeamAgent, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + RedTeamAgent, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = RedTeamAgent( attack_scoring_config=AttackScoringConfig(objective_scorer=mock_objective_scorer), @@ -278,7 +296,10 @@ async def test_normalize_difficult_strategies( ): """Test that DIFFICULT strategy expands to difficult attack strategies.""" with patch.object( - RedTeamAgent, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + RedTeamAgent, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): # DIFFICULT strategy includes TAP which requires FloatScaleThresholdScorer scenario = RedTeamAgent( @@ -298,7 +319,10 @@ async def test_normalize_mixed_difficulty_levels( ): """Test that multiple difficulty levels expand correctly.""" with patch.object( - RedTeamAgent, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + RedTeamAgent, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = RedTeamAgent( attack_scoring_config=AttackScoringConfig(objective_scorer=mock_objective_scorer), @@ -317,7 +341,10 @@ async def test_normalize_with_specific_and_difficulty_levels( ): """Test that specific strategies combined with difficulty levels work correctly.""" with patch.object( - RedTeamAgent, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + RedTeamAgent, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = RedTeamAgent( attack_scoring_config=AttackScoringConfig(objective_scorer=mock_objective_scorer), @@ -344,7 +371,10 @@ async def test_get_attack_from_single_turn_strategy( ): """Test creating an attack from a single-turn strategy.""" with patch.object( - RedTeamAgent, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + RedTeamAgent, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = RedTeamAgent( attack_scoring_config=AttackScoringConfig(objective_scorer=mock_objective_scorer), @@ -358,7 +388,9 @@ async def test_get_attack_from_single_turn_strategy( # Get the composite strategy that was created during initialization composite_strategy = scenario._scenario_composites[0] - atomic_attack = scenario._get_attack_from_strategy(composite_strategy) + atomic_attack = scenario._get_attack_from_strategy( + composite=composite_strategy, seed_groups=mock_memory_seed_groups + ) assert isinstance(atomic_attack, AtomicAttack) assert atomic_attack.seed_groups == mock_memory_seed_groups @@ -373,7 +405,10 @@ async def test_get_attack_from_multi_turn_strategy( ): """Test creating a multi-turn attack strategy.""" with patch.object( - RedTeamAgent, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + RedTeamAgent, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = RedTeamAgent( adversarial_chat=mock_adversarial_target, @@ -388,7 +423,9 @@ async def test_get_attack_from_multi_turn_strategy( # Get the composite strategy that was created during initialization composite_strategy = scenario._scenario_composites[0] - atomic_attack = scenario._get_attack_from_strategy(composite_strategy) + atomic_attack = scenario._get_attack_from_strategy( + composite=composite_strategy, seed_groups=mock_memory_seed_groups + ) assert isinstance(atomic_attack, AtomicAttack) assert atomic_attack.seed_groups == mock_memory_seed_groups @@ -403,7 +440,10 @@ async def test_get_attack_single_turn_with_converters( ): """Test creating a single-turn attack with converters.""" with patch.object( - RedTeamAgent, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + RedTeamAgent, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = RedTeamAgent( attack_scoring_config=AttackScoringConfig(objective_scorer=mock_objective_scorer), @@ -432,7 +472,10 @@ async def test_get_attack_multi_turn_with_adversarial_target( ): """Test creating a multi-turn attack.""" with patch.object( - RedTeamAgent, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + RedTeamAgent, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = RedTeamAgent( adversarial_chat=mock_adversarial_target, @@ -488,7 +531,10 @@ async def test_all_single_turn_strategies_create_attack_runs( ): """Test that all single-turn strategies can create attack runs.""" with patch.object( - RedTeamAgent, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + RedTeamAgent, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = RedTeamAgent( attack_scoring_config=AttackScoringConfig(objective_scorer=mock_objective_scorer), @@ -502,7 +548,9 @@ async def test_all_single_turn_strategies_create_attack_runs( # Get the composite strategy that was created during initialization composite_strategy = scenario._scenario_composites[0] - atomic_attack = scenario._get_attack_from_strategy(composite_strategy) + atomic_attack = scenario._get_attack_from_strategy( + composite=composite_strategy, seed_groups=mock_memory_seed_groups + ) assert isinstance(atomic_attack, AtomicAttack) @pytest.mark.parametrize( @@ -523,7 +571,10 @@ async def test_all_multi_turn_strategies_create_attack_runs( ): """Test that all multi-turn strategies can create attack runs.""" with patch.object( - RedTeamAgent, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + RedTeamAgent, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = RedTeamAgent( adversarial_chat=mock_adversarial_target, @@ -538,7 +589,9 @@ async def test_all_multi_turn_strategies_create_attack_runs( # Get the composite strategy that was created during initialization composite_strategy = scenario._scenario_composites[0] - atomic_attack = scenario._get_attack_from_strategy(composite_strategy) + atomic_attack = scenario._get_attack_from_strategy( + composite=composite_strategy, seed_groups=mock_memory_seed_groups + ) assert isinstance(atomic_attack, AtomicAttack) @@ -553,7 +606,10 @@ async def test_scenario_composites_set_after_initialize( strategies = [FoundryStrategy.Base64, FoundryStrategy.ROT13] with patch.object( - RedTeamAgent, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + RedTeamAgent, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = RedTeamAgent( attack_scoring_config=AttackScoringConfig(objective_scorer=mock_objective_scorer), @@ -592,7 +648,10 @@ async def test_scenario_atomic_attack_count_matches_strategies( ] with patch.object( - RedTeamAgent, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + RedTeamAgent, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = RedTeamAgent( attack_scoring_config=AttackScoringConfig(objective_scorer=mock_objective_scorer), @@ -613,7 +672,10 @@ async def test_initialize_with_foundry_composite_directly( composite = FoundryComposite(attack=FoundryStrategy.Crescendo, converters=[FoundryStrategy.Base64]) with patch.object( - RedTeamAgent, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + RedTeamAgent, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = RedTeamAgent( attack_scoring_config=AttackScoringConfig(objective_scorer=mock_objective_scorer), @@ -638,7 +700,10 @@ async def test_initialize_with_mixed_composites_and_strategies( composite = FoundryComposite(attack=FoundryStrategy.Crescendo, converters=[FoundryStrategy.Base64]) with patch.object( - RedTeamAgent, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + RedTeamAgent, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = RedTeamAgent( attack_scoring_config=AttackScoringConfig(objective_scorer=mock_objective_scorer), @@ -663,7 +728,10 @@ async def test_initialize_converts_scenario_composite_strategy_to_foundry_compos legacy = ScenarioCompositeStrategy(strategies=[FoundryStrategy.Crescendo, FoundryStrategy.Base64]) with patch.object( - RedTeamAgent, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + RedTeamAgent, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = RedTeamAgent( attack_scoring_config=AttackScoringConfig(objective_scorer=mock_objective_scorer), @@ -688,7 +756,10 @@ async def test_initialize_converts_converter_first_composite_strategy( legacy = ScenarioCompositeStrategy(strategies=[FoundryStrategy.Base64, FoundryStrategy.Crescendo]) with patch.object( - RedTeamAgent, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + RedTeamAgent, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = RedTeamAgent( attack_scoring_config=AttackScoringConfig(objective_scorer=mock_objective_scorer), @@ -712,7 +783,10 @@ async def test_initialize_converts_converter_only_composite_strategy( legacy = ScenarioCompositeStrategy(strategies=[FoundryStrategy.Base64, FoundryStrategy.ROT13]) with patch.object( - RedTeamAgent, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_memory_seed_groups + RedTeamAgent, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_memory_seed_groups}, ): scenario = RedTeamAgent( attack_scoring_config=AttackScoringConfig(objective_scorer=mock_objective_scorer), @@ -739,8 +813,8 @@ async def test_one_resolution_call_baseline_matches_strategies(self, mock_object seed_groups = [SeedAttackGroup(seeds=[SeedObjective(value=f"obj{i}")]) for i in range(10)] config = DatasetAttackConfiguration(seed_groups=seed_groups, max_dataset_size=3) - first_sample = seed_groups[:3] - second_sample = seed_groups[5:8] + first_sample = [("inline", group) for group in seed_groups[:3]] + second_sample = [("inline", group) for group in seed_groups[5:8]] with patch( "pyrit.scenario.core.dataset_configuration.random.sample", side_effect=[first_sample, second_sample], diff --git a/tests/unit/scenario/garak/test_encoding.py b/tests/unit/scenario/garak/test_encoding.py index c26784f952..555e4eaf98 100644 --- a/tests/unit/scenario/garak/test_encoding.py +++ b/tests/unit/scenario/garak/test_encoding.py @@ -101,7 +101,7 @@ def test_init_with_default_seed_prompts(self, mock_objective_target, mock_object """Test initialization with default seed prompts (Garak dataset).""" from unittest.mock import patch - with patch.object(Encoding, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=[]): + with patch.object(Encoding, "_resolve_seed_groups_by_dataset_async", new_callable=AsyncMock, return_value={}): scenario = Encoding( objective_scorer=mock_objective_scorer, ) @@ -113,7 +113,7 @@ def test_init_with_custom_scorer(self, mock_objective_target, mock_objective_sco """Test initialization with custom objective scorer.""" from unittest.mock import patch - with patch.object(Encoding, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=[]): + with patch.object(Encoding, "_resolve_seed_groups_by_dataset_async", new_callable=AsyncMock, return_value={}): scenario = Encoding( objective_scorer=mock_objective_scorer, ) @@ -124,7 +124,7 @@ def test_init_creates_default_scorer_when_not_provided(self, mock_objective_targ """Test that initialization creates default DecodingScorer when not provided.""" from unittest.mock import patch - with patch.object(Encoding, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=[]): + with patch.object(Encoding, "_resolve_seed_groups_by_dataset_async", new_callable=AsyncMock, return_value={}): scenario = Encoding() # Should create a DecodingScorer by default @@ -137,7 +137,7 @@ async def test_init_raises_exception_when_no_datasets_available(self, mock_objec from pyrit.scenario.core.dataset_configuration import DatasetConstraintError - # Don't mock _resolve_seed_groups_async; let it try to load from empty memory. + # Don't mock _resolve_seed_groups_by_dataset_async; let it try to load from empty memory. # Disable the provider fallback so memory stays empty and the scenario raises. scenario = Encoding(objective_scorer=mock_objective_scorer) @@ -150,7 +150,7 @@ def test_init_with_memory_labels(self, mock_objective_target, mock_objective_sco """Test initialization with memory labels.""" from unittest.mock import patch - with patch.object(Encoding, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=[]): + with patch.object(Encoding, "_resolve_seed_groups_by_dataset_async", new_callable=AsyncMock, return_value={}): scenario = Encoding( objective_scorer=mock_objective_scorer, ) @@ -164,7 +164,7 @@ def test_init_with_custom_encoding_templates(self, mock_objective_target, mock_o custom_templates = ["template1", "template2"] - with patch.object(Encoding, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=[]): + with patch.object(Encoding, "_resolve_seed_groups_by_dataset_async", new_callable=AsyncMock, return_value={}): scenario = Encoding( encoding_templates=custom_templates, objective_scorer=mock_objective_scorer, @@ -176,7 +176,7 @@ def test_init_with_max_concurrency(self, mock_objective_target, mock_objective_s """Test initialization with custom max_concurrency.""" from unittest.mock import patch - with patch.object(Encoding, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=[]): + with patch.object(Encoding, "_resolve_seed_groups_by_dataset_async", new_callable=AsyncMock, return_value={}): scenario = Encoding( objective_scorer=mock_objective_scorer, ) @@ -191,7 +191,10 @@ async def test_init_attack_strategies( from unittest.mock import patch with patch.object( - Encoding, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_seed_attack_groups + Encoding, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_seed_attack_groups}, ): scenario = Encoding( objective_scorer=mock_objective_scorer, @@ -218,7 +221,10 @@ async def test_get_atomic_attacks_async_returns_attacks( from unittest.mock import patch with patch.object( - Encoding, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_seed_attack_groups + Encoding, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_seed_attack_groups}, ): scenario = Encoding( objective_scorer=mock_objective_scorer, @@ -238,14 +244,17 @@ async def test_get_converter_attacks_returns_multiple_encodings( from unittest.mock import patch with patch.object( - Encoding, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_seed_attack_groups + Encoding, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_seed_attack_groups}, ): scenario = Encoding( objective_scorer=mock_objective_scorer, ) await scenario.initialize_async(objective_target=mock_objective_target, dataset_config=mock_dataset_config) - attack_runs = scenario._get_converter_attacks() + attack_runs = scenario._get_converter_attacks(seed_groups=mock_seed_attack_groups) # Should have multiple attack runs for different encodings # The list includes: Base64 (4 variants), Base2048, Base16, Base32, ASCII85 (2), hex, @@ -259,14 +268,19 @@ async def test_get_prompt_attacks_creates_attack_runs( from unittest.mock import patch with patch.object( - Encoding, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_seed_attack_groups + Encoding, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_seed_attack_groups}, ): scenario = Encoding( objective_scorer=mock_objective_scorer, ) await scenario.initialize_async(objective_target=mock_objective_target, dataset_config=mock_dataset_config) - attack_runs = scenario._get_prompt_attacks(converters=[Base64Converter()], encoding_name="Base64") + attack_runs = scenario._get_prompt_attacks( + converters=[Base64Converter()], encoding_name="Base64", seed_groups=mock_seed_attack_groups + ) # Should create attack runs assert len(attack_runs) > 0 @@ -288,14 +302,19 @@ async def test_attack_runs_include_objectives( from unittest.mock import patch with patch.object( - Encoding, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_seed_attack_groups + Encoding, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_seed_attack_groups}, ): scenario = Encoding( objective_scorer=mock_objective_scorer, ) await scenario.initialize_async(objective_target=mock_objective_target, dataset_config=mock_dataset_config) - attack_runs = scenario._get_prompt_attacks(converters=[Base64Converter()], encoding_name="Base64") + attack_runs = scenario._get_prompt_attacks( + converters=[Base64Converter()], encoding_name="Base64", seed_groups=mock_seed_attack_groups + ) # Check that seed groups contain objectives with the expected format for run in attack_runs: @@ -319,7 +338,10 @@ async def test_scenario_initialization( from unittest.mock import patch with patch.object( - Encoding, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_seed_attack_groups + Encoding, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_seed_attack_groups}, ): scenario = Encoding( objective_scorer=mock_objective_scorer, @@ -333,22 +355,26 @@ async def test_scenario_initialization( async def test_resolve_seed_groups_loads_garak_data( self, mock_objective_target, mock_objective_scorer, mock_seed_attack_groups, mock_dataset_config ): - """Test that _resolve_seed_groups_async loads data from Garak datasets.""" + """Test that _resolve_seed_groups_by_dataset_async loads data from Garak datasets.""" from unittest.mock import patch with patch.object( - Encoding, "_resolve_seed_groups_async", new_callable=AsyncMock, return_value=mock_seed_attack_groups + Encoding, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_seed_attack_groups}, ): scenario = Encoding( objective_scorer=mock_objective_scorer, ) - # After resolve, should have seed groups - resolved = await scenario._resolve_seed_groups_async() - assert len(resolved) > 0 + # After resolve, should have seed groups keyed by dataset + resolved = await scenario._resolve_seed_groups_by_dataset_async() + flattened = [group for groups in resolved.values() for group in groups] + assert flattened # Verify it's returning SeedAttackGroup objects - assert all(isinstance(group, SeedAttackGroup) for group in resolved) + assert all(isinstance(group, SeedAttackGroup) for group in flattened) @pytest.mark.usefixtures("patch_central_database") @@ -451,8 +477,8 @@ async def test_one_resolution_call_baseline_matches_strategies(self, mock_object seed_groups = [SeedAttackGroup(seeds=[SeedObjective(value=f"obj{i}")]) for i in range(10)] config = DatasetAttackConfiguration(seed_groups=seed_groups, max_dataset_size=3) - first_sample = seed_groups[:3] - second_sample = seed_groups[5:8] + first_sample = [("inline", group) for group in seed_groups[:3]] + second_sample = [("inline", group) for group in seed_groups[5:8]] with patch( "pyrit.scenario.core.dataset_configuration.random.sample", side_effect=[first_sample, second_sample],