From a01a85365dd7da8249a2882d15abaa8fc791ac55 Mon Sep 17 00:00:00 2001 From: Richard Lundeen Date: Thu, 16 Apr 2026 16:18:43 -0700 Subject: [PATCH 1/8] refactoring scenario strategy selection --- pyrit/scenario/core/dataset_configuration.py | 10 +- pyrit/scenario/core/scenario.py | 32 ++++- pyrit/scenario/core/scenario_strategy.py | 116 +++++++++++++-- .../scenario/scenarios/airt/content_harms.py | 13 +- pyrit/scenario/scenarios/airt/cyber.py | 9 +- pyrit/scenario/scenarios/airt/jailbreak.py | 6 +- pyrit/scenario/scenarios/airt/leakage.py | 9 +- pyrit/scenario/scenarios/airt/psychosocial.py | 11 +- pyrit/scenario/scenarios/airt/scam.py | 9 +- .../scenarios/foundry/red_team_agent.py | 135 +++++++++--------- pyrit/scenario/scenarios/garak/encoding.py | 9 +- tests/unit/scenario/test_content_harms.py | 18 +-- .../scenario/test_dataset_configuration.py | 12 +- tests/unit/scenario/test_encoding.py | 14 +- .../unit/scenario/test_strategy_validation.py | 48 ++++--- 15 files changed, 261 insertions(+), 190 deletions(-) diff --git a/pyrit/scenario/core/dataset_configuration.py b/pyrit/scenario/core/dataset_configuration.py index 2b78ced2f9..25cd9162c3 100644 --- a/pyrit/scenario/core/dataset_configuration.py +++ b/pyrit/scenario/core/dataset_configuration.py @@ -20,7 +20,7 @@ from collections.abc import Sequence from pyrit.models.seeds.seed import Seed - from pyrit.scenario.core.scenario_strategy import ScenarioCompositeStrategy + from pyrit.scenario.core.scenario_strategy import ScenarioStrategy # Key used when seed_groups are provided directly (not from a named dataset) EXPLICIT_SEED_GROUPS_KEY = "_explicit_seed_groups" @@ -38,7 +38,7 @@ class DatasetConfiguration: dataset_names (Optional[List[str]]): Names of datasets to load from memory. max_dataset_size (Optional[int]): If set, randomly samples up to this many SeedGroups from the configured dataset source (without replacement, so no duplicates). - scenario_composites (Optional[Sequence[ScenarioCompositeStrategy]]): The scenario + scenario_strategies (Optional[Sequence[ScenarioStrategy]]): The scenario strategies being executed. Subclasses can use this to filter or customize which seed groups are loaded based on the selected strategies. """ @@ -49,7 +49,7 @@ def __init__( seed_groups: Optional[list[SeedGroup]] = None, dataset_names: Optional[list[str]] = None, max_dataset_size: Optional[int] = None, - scenario_composites: Optional[Sequence[ScenarioCompositeStrategy]] = None, + scenario_strategies: Optional[Sequence[ScenarioStrategy]] = None, ) -> None: """ Initialize a DatasetConfiguration. @@ -59,7 +59,7 @@ def __init__( dataset_names (Optional[List[str]]): Names of datasets to load from memory. max_dataset_size (Optional[int]): If set, randomly samples up to this many SeedGroups (without replacement). - scenario_composites (Optional[Sequence[ScenarioCompositeStrategy]]): The scenario + scenario_strategies (Optional[Sequence[ScenarioStrategy]]): The scenario strategies being executed. Subclasses can use this to filter or customize which seed groups are loaded. @@ -82,7 +82,7 @@ def __init__( self._seed_groups = list(seed_groups) if seed_groups is not None else None self.max_dataset_size = max_dataset_size self._dataset_names = list(dataset_names) if dataset_names is not None else None - self._scenario_composites = scenario_composites + self._scenario_strategies = scenario_strategies def get_seed_groups(self) -> dict[str, list[SeedGroup]]: """ diff --git a/pyrit/scenario/core/scenario.py b/pyrit/scenario/core/scenario.py index 7e91c53bda..c9483e0cc5 100644 --- a/pyrit/scenario/core/scenario.py +++ b/pyrit/scenario/core/scenario.py @@ -114,8 +114,8 @@ def __init__( self._include_baseline = include_default_baseline - # Store prepared strategy composites for use in _get_atomic_attacks_async - self._scenario_composites: list[ScenarioCompositeStrategy] = [] + # Store prepared strategies for use in _get_atomic_attacks_async + self._scenario_strategies: list[ScenarioStrategy] = [] # Store original objectives for each atomic attack (before any mutations) # Key: atomic_attack_name, Value: tuple of original objectives @@ -186,6 +186,28 @@ def _get_default_objective_scorer(self) -> TrueFalseScorer: logger.info(f"No registered default objective scorer found, using fallback: {type(scorer).__name__}") return scorer + def _prepare_strategies( + self, + strategies: Optional[Sequence[ScenarioStrategy | ScenarioCompositeStrategy]], + ) -> list[ScenarioStrategy]: + """ + Resolve strategy inputs into a concrete list for this scenario. + + The default implementation calls resolve() on the strategy class, which handles + None (use default), empty list (baseline-only), and aggregate expansion. + + Subclasses with complex composition semantics (e.g., RedTeamAgent with + FoundryComposite) should override this to build their own composite types. + + Args: + strategies: Strategy inputs from initialize_async. None means use default, + [] means baseline-only, otherwise a list of strategies to resolve. + + Returns: + list[ScenarioStrategy]: Ordered, deduplicated concrete strategies. + """ + return self._strategy_class.resolve(strategies, default=self.get_default_strategy()) + @apply_defaults async def initialize_async( self, @@ -246,11 +268,7 @@ async def initialize_async( self._memory_labels = memory_labels or {} # Prepare scenario strategies using the stored configuration - # Allow empty strategies when include_baseline is True (baseline-only execution) - self._scenario_composites = self._strategy_class.prepare_scenario_strategies( - scenario_strategies, - default_aggregate=self.get_default_strategy(), - ) + self._scenario_strategies = self._prepare_strategies(scenario_strategies) self._atomic_attacks = await self._get_atomic_attacks_async() diff --git a/pyrit/scenario/core/scenario_strategy.py b/pyrit/scenario/core/scenario_strategy.py index 0252f68415..ccbc0f6c67 100644 --- a/pyrit/scenario/core/scenario_strategy.py +++ b/pyrit/scenario/core/scenario_strategy.py @@ -202,28 +202,78 @@ def normalize_strategies(cls: type[T], strategies: set[T]) -> set[T]: Set[T]: The normalized set of concrete attack strategies with aggregate tags expanded and removed. """ - normalized_strategies = set(strategies) + print_deprecation_message( + old_item="ScenarioStrategy.normalize_strategies", + new_item="ScenarioStrategy.expand", + removed_in="0.8.0", + ) + return set(cls.expand(strategies)) - # Find aggregate tags in the input and expand them + @classmethod + def expand(cls: type[T], strategies: set[T]) -> list[T]: + """ + Expand a set of strategies (including aggregates) into an ordered, deduplicated list. + + Aggregate markers (like EASY, ALL) are expanded into their constituent concrete strategies. + The result is sorted by enum definition order for determinism. + + Args: + strategies (set[T]): Set of strategies, which may include aggregate markers. + + Returns: + list[T]: Ordered list of concrete strategies with aggregates expanded. + """ + concrete: set[T] = set(strategies) aggregate_tags = cls.get_aggregate_tags() aggregates_to_expand = { tag for strategy in strategies if strategy.value in aggregate_tags for tag in strategy.tags } - for aggregate_tag in aggregates_to_expand: - # Remove the aggregate marker itself - aggregate_marker = next((s for s in normalized_strategies if s.value == aggregate_tag), None) + aggregate_marker = next((s for s in concrete if s.value == aggregate_tag), None) if aggregate_marker: - normalized_strategies.remove(aggregate_marker) - - # Special handling for "all" tag - expand to all non-aggregate strategies + concrete.remove(aggregate_marker) if aggregate_tag == "all": - normalized_strategies.update(cls.get_all_strategies()) + concrete.update(cls.get_all_strategies()) else: - # Add all strategies with that tag - normalized_strategies.update(cls.get_strategies_by_tag(aggregate_tag)) + concrete.update(cls.get_strategies_by_tag(aggregate_tag)) + return [s for s in cls if s in concrete] - return normalized_strategies + @classmethod + def resolve(cls: type[T], strategies: Sequence[Any] | None, *, default: T) -> list[T]: + """ + Resolve strategy inputs into a concrete, ordered, deduplicated list. + + Handles None (returns expanded default), plain strategies, and aggregate strategies. + Non-cls items (e.g., ScenarioCompositeStrategy) are silently skipped for + backward compatibility. + + Args: + strategies (Sequence[Any] | None): Strategies to resolve. If None, expands the + default. If an empty list, returns empty (baseline-only execution). + default (T): Default aggregate strategy to use when strategies is None. + + Returns: + list[T]: Ordered, deduplicated list of concrete strategies. + """ + if strategies is None: + return cls.expand({default}) + + result: list[T] = [] + seen: set[T] = set() + aggregate_tags = cls.get_aggregate_tags() + for item in strategies: + if not isinstance(item, cls): + continue + if item.value in aggregate_tags: + for s in cls.expand({item}): + if s not in seen: + seen.add(s) + result.append(s) + else: + if item not in seen: + seen.add(item) + result.append(item) + return result @classmethod def prepare_scenario_strategies( @@ -260,6 +310,11 @@ def prepare_scenario_strategies( ValueError: If strategies is None and default_aggregate is None, or if compositions are invalid according to validate_composition(). """ + print_deprecation_message( + old_item="ScenarioStrategy.prepare_scenario_strategies", + new_item="ScenarioStrategy.resolve and Scenario._prepare_strategies", + removed_in="0.8.0", + ) # Handle None input with default aggregate if strategies is None: if default_aggregate is None: @@ -311,6 +366,11 @@ def supports_composition(cls: type[T]) -> bool: Returns: bool: True if composition is supported, False otherwise. """ + print_deprecation_message( + old_item="ScenarioStrategy.supports_composition", + new_item="FoundryComposite dataclass", + removed_in="0.8.0", + ) return False @classmethod @@ -340,6 +400,11 @@ def validate_composition(cls: type[T], strategies: Sequence[T]) -> None: >>> FoundryStrategy.validate_composition([FoundryStrategy.Crescendo, FoundryStrategy.MultiTurn]) ValueError: Cannot compose multiple attack strategies: ['crescendo', 'multi_turn'] """ + print_deprecation_message( + old_item="ScenarioStrategy.validate_composition", + new_item="FoundryComposite dataclass", + removed_in="0.8.0", + ) if not strategies: raise ValueError("Cannot validate empty strategy list") @@ -409,7 +474,11 @@ def __init__(self, *, strategies: Sequence[ScenarioStrategy]) -> None: raise ValueError("strategies list cannot be empty") self._strategies = list(strategies) - self._name = self.get_composite_name(self._strategies) + if len(self._strategies) == 1: + self._name = str(self._strategies[0].value) + else: + strategy_names = ", ".join(s.value for s in self._strategies) + self._name = f"ComposedStrategy({strategy_names})" @property def name(self) -> str: @@ -451,6 +520,13 @@ def extract_single_strategy_values( Raises: ValueError: If any composite contains multiple strategies. """ + from pyrit.common.deprecation import print_deprecation_message + + print_deprecation_message( + old_item="ScenarioCompositeStrategy.extract_single_strategy_values", + new_item="[s.value for s in self._scenario_strategies]", + removed_in="0.8.0", + ) # Check that all composites are single-strategy multi_strategy_composites = [comp for comp in composites if not comp.is_single_strategy] if multi_strategy_composites: @@ -496,6 +572,13 @@ def get_composite_name(strategies: Sequence[ScenarioStrategy]) -> str: ... ]) >>> # Returns: "ComposedStrategy(base64, atbash)" """ + from pyrit.common.deprecation import print_deprecation_message + + print_deprecation_message( + old_item="ScenarioCompositeStrategy.get_composite_name", + new_item="ScenarioCompositeStrategy.name property", + removed_in="0.8.0", + ) if not strategies: raise ValueError("Cannot generate name for empty strategy list") @@ -544,6 +627,13 @@ def normalize_compositions( # Error: Cannot mix aggregate with concrete in same composition [ScenarioCompositeStrategy(strategies=[EASY, Base64])] -> ValueError """ + from pyrit.common.deprecation import print_deprecation_message + + print_deprecation_message( + old_item="ScenarioCompositeStrategy.normalize_compositions", + new_item="ScenarioStrategy.resolve and Scenario._prepare_strategies", + removed_in="0.8.0", + ) if not compositions: raise ValueError("Compositions list cannot be empty") diff --git a/pyrit/scenario/scenarios/airt/content_harms.py b/pyrit/scenario/scenarios/airt/content_harms.py index d22ece85ff..400e19667d 100644 --- a/pyrit/scenario/scenarios/airt/content_harms.py +++ b/pyrit/scenario/scenarios/airt/content_harms.py @@ -24,10 +24,7 @@ from pyrit.scenario.core.attack_technique import AttackTechnique from pyrit.scenario.core.dataset_configuration import DatasetConfiguration from pyrit.scenario.core.scenario import Scenario -from pyrit.scenario.core.scenario_strategy import ( - ScenarioCompositeStrategy, - ScenarioStrategy, -) +from pyrit.scenario.core.scenario_strategy import ScenarioStrategy from pyrit.score import TrueFalseScorer logger = logging.getLogger(__name__) @@ -57,13 +54,11 @@ def get_seed_groups(self) -> dict[str, list[SeedGroup]]: """ result = super().get_seed_groups() - if self._scenario_composites is None: + if self._scenario_strategies is None: return result # Extract selected harm strategies - selected_harms = ScenarioCompositeStrategy.extract_single_strategy_values( - self._scenario_composites, strategy_type=ContentHarmsStrategy - ) + selected_harms = {s.value for s in self._scenario_strategies if isinstance(s, ContentHarmsStrategy)} # Filter to matching datasets and map keys to harm names mapped_result: dict[str, list[SeedGroup]] = {} @@ -201,7 +196,7 @@ def _resolve_seed_groups_by_harm(self) -> dict[str, list[SeedAttackGroup]]: seed attack groups. """ # Set scenario_composites on the config so get_seed_attack_groups can filter by strategy - self._dataset_config._scenario_composites = self._scenario_composites + self._dataset_config._scenario_strategies = self._scenario_strategies return self._dataset_config.get_seed_attack_groups() async def _get_atomic_attacks_async(self) -> list[AtomicAttack]: diff --git a/pyrit/scenario/scenarios/airt/cyber.py b/pyrit/scenario/scenarios/airt/cyber.py index fe53b62bb9..57b6a3edf8 100644 --- a/pyrit/scenario/scenarios/airt/cyber.py +++ b/pyrit/scenario/scenarios/airt/cyber.py @@ -20,10 +20,7 @@ from pyrit.scenario.core.attack_technique import AttackTechnique from pyrit.scenario.core.dataset_configuration import DatasetConfiguration from pyrit.scenario.core.scenario import Scenario -from pyrit.scenario.core.scenario_strategy import ( - ScenarioCompositeStrategy, - ScenarioStrategy, -) +from pyrit.scenario.core.scenario_strategy import ScenarioStrategy from pyrit.score import ( SelfAskRefusalScorer, SelfAskTrueFalseScorer, @@ -265,8 +262,6 @@ async def _get_atomic_attacks_async(self) -> list[AtomicAttack]: # Resolve seed groups from deprecated objectives or dataset config self._seed_groups = self._resolve_seed_groups() - strategies = ScenarioCompositeStrategy.extract_single_strategy_values( - composites=self._scenario_composites, strategy_type=CyberStrategy - ) + strategies = {s.value for s in self._scenario_strategies} return [self._get_atomic_attack_from_strategy(strategy) for strategy in strategies] diff --git a/pyrit/scenario/scenarios/airt/jailbreak.py b/pyrit/scenario/scenarios/airt/jailbreak.py index 9667cca820..8dfe49021c 100644 --- a/pyrit/scenario/scenarios/airt/jailbreak.py +++ b/pyrit/scenario/scenarios/airt/jailbreak.py @@ -25,7 +25,7 @@ from pyrit.scenario.core.attack_technique import AttackTechnique from pyrit.scenario.core.dataset_configuration import DatasetConfiguration from pyrit.scenario.core.scenario import Scenario -from pyrit.scenario.core.scenario_strategy import ScenarioCompositeStrategy, ScenarioStrategy +from pyrit.scenario.core.scenario_strategy import ScenarioStrategy from pyrit.score import ( TrueFalseScorer, ) @@ -315,9 +315,7 @@ async def _get_atomic_attacks_async(self) -> list[AtomicAttack]: # Retrieve seed prompts based on selected strategies self._seed_groups = self._resolve_seed_groups() - strategies = ScenarioCompositeStrategy.extract_single_strategy_values( - composites=self._scenario_composites, strategy_type=JailbreakStrategy - ) + strategies = {s.value for s in self._scenario_strategies} for strategy in strategies: for template_name in self._jailbreaks: diff --git a/pyrit/scenario/scenarios/airt/leakage.py b/pyrit/scenario/scenarios/airt/leakage.py index 43354690c5..41877e9ef0 100644 --- a/pyrit/scenario/scenarios/airt/leakage.py +++ b/pyrit/scenario/scenarios/airt/leakage.py @@ -27,10 +27,7 @@ from pyrit.scenario.core.attack_technique import AttackTechnique from pyrit.scenario.core.dataset_configuration import DatasetConfiguration from pyrit.scenario.core.scenario import Scenario -from pyrit.scenario.core.scenario_strategy import ( - ScenarioCompositeStrategy, - ScenarioStrategy, -) +from pyrit.scenario.core.scenario_strategy import ScenarioStrategy from pyrit.score import ( SelfAskRefusalScorer, SelfAskTrueFalseScorer, @@ -376,8 +373,6 @@ async def _get_atomic_attacks_async(self) -> list[AtomicAttack]: # Resolve objectives to seed groups format self._seed_groups = self._resolve_seed_groups() - strategies = ScenarioCompositeStrategy.extract_single_strategy_values( - composites=self._scenario_composites, strategy_type=LeakageStrategy - ) + strategies = {s.value for s in self._scenario_strategies} return [await self._get_atomic_attack_from_strategy_async(strategy) for strategy in strategies] diff --git a/pyrit/scenario/scenarios/airt/psychosocial.py b/pyrit/scenario/scenarios/airt/psychosocial.py index 2caf2d6f78..9d201baf9d 100644 --- a/pyrit/scenario/scenarios/airt/psychosocial.py +++ b/pyrit/scenario/scenarios/airt/psychosocial.py @@ -315,12 +315,11 @@ def _extract_harm_category_filter(self) -> Optional[str]: Returns: Optional[str]: The harm category to filter by, or None if no filter is set. """ - for composite in self._scenario_composites: - for strategy in composite.strategies: - if isinstance(strategy, PsychosocialStrategy): - harm_filter = strategy.harm_category_filter - if harm_filter: - return harm_filter + for strategy in self._scenario_strategies: + if isinstance(strategy, PsychosocialStrategy): + harm_filter = strategy.harm_category_filter + if harm_filter: + return harm_filter return None def _filter_by_harm_category( diff --git a/pyrit/scenario/scenarios/airt/scam.py b/pyrit/scenario/scenarios/airt/scam.py index e2db719f40..e1d70457af 100644 --- a/pyrit/scenario/scenarios/airt/scam.py +++ b/pyrit/scenario/scenarios/airt/scam.py @@ -28,10 +28,7 @@ from pyrit.scenario.core.attack_technique import AttackTechnique from pyrit.scenario.core.dataset_configuration import DatasetConfiguration from pyrit.scenario.core.scenario import Scenario -from pyrit.scenario.core.scenario_strategy import ( - ScenarioCompositeStrategy, - ScenarioStrategy, -) +from pyrit.scenario.core.scenario_strategy import ScenarioStrategy from pyrit.score import ( SelfAskRefusalScorer, SelfAskTrueFalseScorer, @@ -304,8 +301,6 @@ async def _get_atomic_attacks_async(self) -> list[AtomicAttack]: # Resolve seed groups from deprecated objectives or dataset config self._seed_groups = self._resolve_seed_groups() - strategies = ScenarioCompositeStrategy.extract_single_strategy_values( - composites=self._scenario_composites, strategy_type=ScamStrategy - ) + strategies = {s.value for s in self._scenario_strategies} return [self._get_atomic_attack_from_strategy(strategy) for strategy in strategies] diff --git a/pyrit/scenario/scenarios/foundry/red_team_agent.py b/pyrit/scenario/scenarios/foundry/red_team_agent.py index 2095827e02..2b9658b650 100644 --- a/pyrit/scenario/scenarios/foundry/red_team_agent.py +++ b/pyrit/scenario/scenarios/foundry/red_team_agent.py @@ -12,6 +12,7 @@ import logging import os from collections.abc import Sequence +from dataclasses import dataclass, field from inspect import signature from typing import TYPE_CHECKING, Any, Optional, TypeVar @@ -65,10 +66,7 @@ from pyrit.scenario.core.attack_technique import AttackTechnique from pyrit.scenario.core.dataset_configuration import DatasetConfiguration from pyrit.scenario.core.scenario import Scenario -from pyrit.scenario.core.scenario_strategy import ( - ScenarioCompositeStrategy, - ScenarioStrategy, -) +from pyrit.scenario.core.scenario_strategy import ScenarioStrategy if TYPE_CHECKING: from pyrit.executor.attack.core.attack_strategy import AttackStrategy @@ -77,6 +75,29 @@ logger = logging.getLogger(__name__) +@dataclass +class FoundryComposite: + """ + A typed composition of Foundry attack strategies. + + Exactly one attack strategy (e.g., Crescendo) paired with zero or more + converter strategies (e.g., Base64, ROT13). When no attack is specified, + a PromptSendingAttack is used. + """ + + attack: "FoundryStrategy | None" + converters: "list[FoundryStrategy]" = field(default_factory=list) + + @property + def name(self) -> str: + """Return a human-readable name for this composite.""" + attack_name = self.attack.value if self.attack else "baseline" + if not self.converters: + return attack_name + converter_names = ", ".join(c.value for c in self.converters) + return f"ComposedStrategy({attack_name}, {converter_names})" + + class FoundryStrategy(ScenarioStrategy): """ Strategies for attacks with tag-based categorization. @@ -154,47 +175,6 @@ def get_aggregate_tags(cls) -> set[str]: # Include base class aggregates ("all") and add Foundry-specific ones return super().get_aggregate_tags() | {"easy", "moderate", "difficult", "converter", "attack"} - @classmethod - def supports_composition(cls) -> bool: - """ - Indicate that FoundryStrategy supports composition. - - Returns: - bool: True, as Foundry strategies can be composed together (with rules). - """ - return True - - @classmethod - def validate_composition(cls, strategies: Sequence[ScenarioStrategy]) -> None: - """ - Validate whether the given Foundry strategies can be composed together. - - Foundry-specific composition rules: - - Multiple attack strategies (e.g., Crescendo, MultiTurn) cannot be composed together - - Converters can be freely composed with each other - - At most one attack can be composed with any number of converters - - Args: - strategies (Sequence[ScenarioStrategy]): The strategies to validate for composition. - - Raises: - ValueError: If the composition violates Foundry's rules (e.g., multiple attack). - """ - if not strategies: - raise ValueError("Cannot validate empty strategy list") - - # Filter to only FoundryStrategy instances - foundry_strategies = [s for s in strategies if isinstance(s, FoundryStrategy)] - - # Foundry-specific rule: Cannot compose multiple attack strategies - attacks = [s for s in foundry_strategies if "attack" in s.tags] - - if len(attacks) > 1: - raise ValueError( - f"Cannot compose multiple attack strategies together: {[a.value for a in attacks]}. " - f"Only one attack strategy is allowed per composition." - ) - class RedTeamAgent(Scenario): """ @@ -285,6 +265,34 @@ def __init__( include_default_baseline=include_baseline, scenario_result_id=scenario_result_id, ) + self._scenario_composites: list[FoundryComposite] = [] + + def _prepare_strategies(self, strategies: Sequence[ScenarioStrategy] | None) -> list[ScenarioStrategy]: + """ + Resolve strategies and build FoundryComposite objects. + + Each resolved FoundryStrategy becomes a separate FoundryComposite. + Strategies tagged "attack" populate composite.attack; strategies tagged + "converter" populate composite.converters as a single-converter composite. + + Args: + strategies: Optional sequence of FoundryStrategy members (or None to use default). + + Returns: + list[ScenarioStrategy]: The resolved strategies stored in self._scenario_strategies. + """ + resolved = FoundryStrategy.resolve(strategies, default=self.get_default_strategy()) + self._scenario_composites = [self._strategy_to_composite(s) for s in resolved] + return list(resolved) + + @staticmethod + def _strategy_to_composite(strategy: ScenarioStrategy) -> "FoundryComposite": + """Wrap a single FoundryStrategy in a FoundryComposite.""" + if not isinstance(strategy, FoundryStrategy): + raise ValueError(f"Expected FoundryStrategy, got {type(strategy)}") + if "attack" in strategy.tags: + return FoundryComposite(attack=strategy) + return FoundryComposite(attack=None, converters=[strategy]) def _resolve_seed_groups(self) -> list[SeedAttackGroup]: """ @@ -316,49 +324,34 @@ def _get_default_adversarial_target(self) -> OpenAIChatTarget: temperature=1.2, ) - def _get_attack_from_strategy(self, composite_strategy: ScenarioCompositeStrategy) -> AtomicAttack: + def _get_attack_from_strategy(self, composite: FoundryComposite) -> AtomicAttack: """ - Get an atomic attack for the specified strategy composition. + Get an atomic attack for the specified FoundryComposite. Args: - composite_strategy (ScenarioCompositeStrategy): Composite strategy containing one or more - FoundryStrategy enum members to compose together. Can include attack strategies - (e.g., Crescendo, MultiTurn) and converter strategies (e.g., Base64, ROT13) that - will be applied to the same prompts. + composite (FoundryComposite): Typed composite with an optional attack strategy + and zero or more converter strategies. Returns: AtomicAttack: The configured atomic attack. - - Raises: - ValueError: If the strategy composition is invalid (e.g., multiple attack strategies). """ attack: AttackStrategy[Any, Any] - # Extract FoundryStrategy enums from the composite - strategy_list = [s for s in composite_strategy.strategies if isinstance(s, FoundryStrategy)] - - attacks = [s for s in strategy_list if "attack" in s.tags] - converters_strategies = [s for s in strategy_list if "converter" in s.tags] - - # Validate attack composition - if len(attacks) > 1: - raise ValueError(f"Cannot compose multiple attack strategies: {[a.value for a in attacks]}") - attack_type: type[AttackStrategy[Any, Any]] = PromptSendingAttack attack_kwargs: dict[str, Any] = {} - if len(attacks) == 1: - if attacks[0] == FoundryStrategy.Crescendo: + if composite.attack is not None: + if composite.attack == FoundryStrategy.Crescendo: attack_type = CrescendoAttack - elif attacks[0] == FoundryStrategy.MultiTurn: + elif composite.attack == FoundryStrategy.MultiTurn: attack_type = RedTeamingAttack - elif attacks[0] == FoundryStrategy.Pair: + elif composite.attack == FoundryStrategy.Pair: attack_type = TreeOfAttacksWithPruningAttack attack_kwargs = {"tree_width": 1} - elif attacks[0] == FoundryStrategy.Tap: + elif composite.attack == FoundryStrategy.Tap: attack_type = TreeOfAttacksWithPruningAttack converters: list[PromptConverter] = [] - for strategy in converters_strategies: + for strategy in composite.converters: if strategy == FoundryStrategy.AnsiAttack: converters.append(AnsiAttackConverter()) elif strategy == FoundryStrategy.AsciiArt: @@ -408,7 +401,7 @@ def _get_attack_from_strategy(self, composite_strategy: ScenarioCompositeStrateg attack = self._get_attack(attack_type=attack_type, converters=converters, attack_kwargs=attack_kwargs) return AtomicAttack( - atomic_attack_name=composite_strategy.name, + atomic_attack_name=composite.name, attack_technique=AttackTechnique(attack=attack), seed_groups=self._seed_groups, adversarial_chat=self._adversarial_chat, diff --git a/pyrit/scenario/scenarios/garak/encoding.py b/pyrit/scenario/scenarios/garak/encoding.py index 1864c2a93e..ccf4550456 100644 --- a/pyrit/scenario/scenarios/garak/encoding.py +++ b/pyrit/scenario/scenarios/garak/encoding.py @@ -36,10 +36,7 @@ from pyrit.scenario.core.attack_technique import AttackTechnique from pyrit.scenario.core.dataset_configuration import DatasetConfiguration from pyrit.scenario.core.scenario import Scenario -from pyrit.scenario.core.scenario_strategy import ( - ScenarioCompositeStrategy, - ScenarioStrategy, -) +from pyrit.scenario.core.scenario_strategy import ScenarioStrategy from pyrit.score import TrueFalseScorer from pyrit.score.true_false.decoding_scorer import DecodingScorer @@ -274,9 +271,7 @@ def _get_converter_attacks(self) -> list[AtomicAttack]: ] # Filter to only include selected strategies - selected_encoding_names = ScenarioCompositeStrategy.extract_single_strategy_values( - self._scenario_composites, strategy_type=EncodingStrategy - ) + selected_encoding_names = {s.value for s in self._scenario_strategies} converters_with_encodings = [ (conv, name) for conv, name in all_converters_with_encodings if name in selected_encoding_names ] diff --git a/tests/unit/scenario/test_content_harms.py b/tests/unit/scenario/test_content_harms.py index 1e177e15bd..b05a18ff6e 100644 --- a/tests/unit/scenario/test_content_harms.py +++ b/tests/unit/scenario/test_content_harms.py @@ -13,7 +13,6 @@ from pyrit.models import SeedAttackGroup, SeedObjective, SeedPrompt from pyrit.prompt_target import PromptTarget from pyrit.prompt_target.common.prompt_chat_target import PromptChatTarget -from pyrit.scenario import ScenarioCompositeStrategy from pyrit.scenario.airt import ( ContentHarms, ContentHarmsStrategy, @@ -276,7 +275,7 @@ async def test_initialization_with_custom_strategies( await scenario.initialize_async(objective_target=mock_objective_target, scenario_strategies=strategies) # Prepared composites should match provided strategies - assert len(scenario._scenario_composites) == 2 + assert len(scenario._scenario_strategies) == 2 def test_initialization_with_custom_scorer( self, mock_objective_target, mock_adversarial_target, mock_objective_scorer @@ -356,7 +355,7 @@ async def test_initialization_defaults_to_all_strategy( await scenario.initialize_async(objective_target=mock_objective_target) # Should have strategies from the ALL aggregate - assert len(scenario._scenario_composites) > 0 + assert len(scenario._scenario_strategies) > 0 def test_get_default_strategy_returns_all(self): """Test that get_default_strategy returns ALL strategy.""" @@ -548,7 +547,7 @@ def test_get_seed_attack_groups_filters_by_selected_harm_strategy(self): config = ContentHarmsDatasetConfiguration( dataset_names=["airt_hate", "airt_violence", "airt_sexual"], - scenario_composites=[ScenarioCompositeStrategy(strategies=[ContentHarmsStrategy.Hate])], + scenario_strategies=[ContentHarmsStrategy.Hate], ) with patch.object(config, "_load_seed_groups_for_dataset") as mock_load: @@ -571,10 +570,7 @@ def test_get_seed_attack_groups_maps_dataset_names_to_harm_names(self): config = ContentHarmsDatasetConfiguration( dataset_names=["airt_hate", "airt_fairness"], - scenario_composites=[ - ScenarioCompositeStrategy(strategies=[ContentHarmsStrategy.Hate]), - ScenarioCompositeStrategy(strategies=[ContentHarmsStrategy.Fairness]), - ], + scenario_strategies=[ContentHarmsStrategy.Hate, ContentHarmsStrategy.Fairness], ) with patch.object(config, "_load_seed_groups_for_dataset") as mock_load: @@ -603,11 +599,11 @@ def test_get_seed_attack_groups_with_all_strategy_returns_all_harms(self): # ALL strategy expands to all individual harm strategies all_harms = ["hate", "fairness", "violence", "sexual", "harassment", "misinformation", "leakage"] - composites = [ScenarioCompositeStrategy(strategies=[ContentHarmsStrategy(harm)]) for harm in all_harms] + composites = [ContentHarmsStrategy(harm) for harm in all_harms] config = ContentHarmsDatasetConfiguration( dataset_names=all_datasets, - scenario_composites=composites, + scenario_strategies=composites, ) with patch.object(config, "_load_seed_groups_for_dataset") as mock_load: @@ -630,7 +626,7 @@ def test_get_seed_attack_groups_applies_max_dataset_size(self): config = ContentHarmsDatasetConfiguration( dataset_names=["airt_hate"], max_dataset_size=2, - scenario_composites=[ScenarioCompositeStrategy(strategies=[ContentHarmsStrategy.Hate])], + scenario_strategies=[ContentHarmsStrategy.Hate], ) with patch.object(config, "_load_seed_groups_for_dataset") as mock_load: diff --git a/tests/unit/scenario/test_dataset_configuration.py b/tests/unit/scenario/test_dataset_configuration.py index 0a1fae4b8f..e1b5c68727 100644 --- a/tests/unit/scenario/test_dataset_configuration.py +++ b/tests/unit/scenario/test_dataset_configuration.py @@ -56,7 +56,7 @@ def test_init_with_seed_groups_only(self, sample_seed_groups: list) -> None: assert config._seed_groups == sample_seed_groups assert config._dataset_names is None assert config.max_dataset_size is None - assert config._scenario_composites is None + assert config._scenario_strategies is None def test_init_with_dataset_names_only(self) -> None: """Test initialization with only dataset_names.""" @@ -119,15 +119,15 @@ def test_init_copies_dataset_names_to_prevent_mutation(self) -> None: # Config should still have the original length assert len(config._dataset_names) == 2 - def test_init_with_scenario_composites(self, sample_seed_groups: list) -> None: - """Test initialization with scenario_composites.""" - mock_composites = [MagicMock(), MagicMock()] + def test_init_with_scenario_strategies(self, sample_seed_groups: list) -> None: + """Test initialization with scenario_strategies.""" + mock_strategies = [MagicMock(), MagicMock()] config = DatasetConfiguration( seed_groups=sample_seed_groups, - scenario_composites=mock_composites, + scenario_strategies=mock_strategies, ) - assert config._scenario_composites == mock_composites + assert config._scenario_strategies == mock_strategies def test_init_with_no_data_source(self) -> None: """Test initialization with neither seed_groups nor dataset_names.""" diff --git a/tests/unit/scenario/test_encoding.py b/tests/unit/scenario/test_encoding.py index 79a802d2ef..1271bdaa82 100644 --- a/tests/unit/scenario/test_encoding.py +++ b/tests/unit/scenario/test_encoding.py @@ -193,17 +193,11 @@ async def test_init_attack_strategies( await scenario.initialize_async(objective_target=mock_objective_target, dataset_config=mock_dataset_config) # By default, EncodingStrategy.ALL is used, which expands to all encoding strategies - assert len(scenario._scenario_composites) > 0 - # Verify all composites contain EncodingStrategy instances - assert all( - isinstance(comp.strategies[0], EncodingStrategy) - for comp in scenario._scenario_composites - if comp.strategies - ) + assert len(scenario._scenario_strategies) > 0 + # Verify all strategies contain EncodingStrategy instances + assert all(isinstance(s, EncodingStrategy) for s in scenario._scenario_strategies) # Verify none of the strategies are the aggregate "ALL" - assert all( - comp.strategies[0] != EncodingStrategy.ALL for comp in scenario._scenario_composites if comp.strategies - ) + assert all(s != EncodingStrategy.ALL for s in scenario._scenario_strategies) @pytest.mark.usefixtures("patch_central_database") diff --git a/tests/unit/scenario/test_strategy_validation.py b/tests/unit/scenario/test_strategy_validation.py index 2721b9ac62..6b8c457af9 100644 --- a/tests/unit/scenario/test_strategy_validation.py +++ b/tests/unit/scenario/test_strategy_validation.py @@ -7,6 +7,7 @@ from pyrit.scenario import ScenarioCompositeStrategy from pyrit.scenario.foundry import FoundryStrategy +from pyrit.scenario.foundry.red_team_agent import FoundryComposite from pyrit.scenario.garak import EncodingStrategy @@ -28,29 +29,36 @@ def test_foundry_validation_allows_single_strategy(self): # Should not raise FoundryStrategy.validate_composition([FoundryStrategy.Base64]) - def test_foundry_validation_allows_converter_composition(self): - """Test that foundry validation allows multiple converters.""" - # Should not raise - FoundryStrategy.validate_composition([FoundryStrategy.Base64, FoundryStrategy.Atbash]) - def test_foundry_validation_allows_one_attack_with_converters(self): - """Test that foundry validation allows one attack with converters.""" - # Should not raise - FoundryStrategy.validate_composition( - [FoundryStrategy.Base64, FoundryStrategy.Crescendo, FoundryStrategy.Atbash] +class TestFoundryComposite: + """Tests for FoundryComposite dataclass construction and naming.""" + + def test_converter_only_composite_name(self): + """Test name for a composite with only a converter strategy.""" + composite = FoundryComposite(attack=None, converters=[FoundryStrategy.Base64]) + assert composite.name == "ComposedStrategy(baseline, base64)" + + def test_attack_only_composite_name(self): + """Test name for a composite with only an attack strategy.""" + composite = FoundryComposite(attack=FoundryStrategy.Crescendo) + assert composite.name == "crescendo" + + def test_attack_with_converter_composite_name(self): + """Test name for attack + converter composition.""" + composite = FoundryComposite(attack=FoundryStrategy.Crescendo, converters=[FoundryStrategy.Base64]) + assert composite.name == "ComposedStrategy(crescendo, base64)" + + def test_attack_with_multiple_converters_composite_name(self): + """Test name with multiple converters.""" + composite = FoundryComposite( + attack=FoundryStrategy.Crescendo, converters=[FoundryStrategy.Base64, FoundryStrategy.Atbash] ) + assert composite.name == "ComposedStrategy(crescendo, base64, atbash)" - def test_foundry_validation_rejects_multiple_attacks(self): - """Test that foundry validation rejects multiple attack strategies.""" - with pytest.raises(ValueError, match="Cannot compose multiple attack strategies"): - FoundryStrategy.validate_composition([FoundryStrategy.Crescendo, FoundryStrategy.MultiTurn]) - - def test_foundry_validation_rejects_attacks_with_converters_and_another_attack(self): - """Test that foundry validation rejects multiple attacks even with converters.""" - with pytest.raises(ValueError, match="Cannot compose multiple attack strategies"): - FoundryStrategy.validate_composition( - [FoundryStrategy.Base64, FoundryStrategy.Crescendo, FoundryStrategy.MultiTurn] - ) + def test_empty_composite_defaults(self): + """Test that FoundryComposite defaults converters to empty list.""" + composite = FoundryComposite(attack=FoundryStrategy.Base64) + assert composite.converters == [] class TestScenarioCompositeStrategyExtraction: From 6e1298eebaa8f00925cc0b118fdd91ca55e5a72d Mon Sep 17 00:00:00 2001 From: Richard Lundeen Date: Thu, 16 Apr 2026 16:54:31 -0700 Subject: [PATCH 2/8] refactor a bit --- pyrit/scenario/core/scenario.py | 15 +- pyrit/scenario/core/scenario_strategy.py | 175 +----------------- pyrit/scenario/scenarios/foundry/__init__.py | 2 + .../scenarios/foundry/red_team_agent.py | 98 ++++++++-- tests/unit/scenario/test_foundry.py | 99 +++++++++- .../unit/scenario/test_strategy_validation.py | 45 ++--- 6 files changed, 223 insertions(+), 211 deletions(-) diff --git a/pyrit/scenario/core/scenario.py b/pyrit/scenario/core/scenario.py index c9483e0cc5..f53f675016 100644 --- a/pyrit/scenario/core/scenario.py +++ b/pyrit/scenario/core/scenario.py @@ -29,10 +29,7 @@ from pyrit.scenario.core.atomic_attack import AtomicAttack from pyrit.scenario.core.attack_technique import AttackTechnique from pyrit.scenario.core.dataset_configuration import DatasetConfiguration -from pyrit.scenario.core.scenario_strategy import ( - ScenarioCompositeStrategy, - ScenarioStrategy, -) +from pyrit.scenario.core.scenario_strategy import ScenarioStrategy from pyrit.score import Scorer, SelfAskRefusalScorer, TrueFalseInverterScorer, TrueFalseScorer if TYPE_CHECKING: @@ -188,7 +185,7 @@ def _get_default_objective_scorer(self) -> TrueFalseScorer: def _prepare_strategies( self, - strategies: Optional[Sequence[ScenarioStrategy | ScenarioCompositeStrategy]], + strategies: Optional[Sequence[ScenarioStrategy]], ) -> list[ScenarioStrategy]: """ Resolve strategy inputs into a concrete list for this scenario. @@ -213,7 +210,7 @@ async def initialize_async( self, *, objective_target: PromptTarget = REQUIRED_VALUE, # type: ignore[assignment] - scenario_strategies: Optional[Sequence[ScenarioStrategy | ScenarioCompositeStrategy]] = None, + scenario_strategies: Optional[Sequence[ScenarioStrategy]] = None, dataset_config: Optional[DatasetConfiguration] = None, max_concurrency: int = 10, max_retries: int = 0, @@ -232,10 +229,8 @@ async def initialize_async( Args: objective_target (PromptTarget): The target system to attack. - scenario_strategies (Optional[Sequence[ScenarioStrategy | ScenarioCompositeStrategy]]): - The strategies to execute. Can be a list of bare ScenarioStrategy enums or - ScenarioCompositeStrategy instances for advanced composition. Bare enums are - automatically wrapped into composites. If None, uses the default aggregate + scenario_strategies (Optional[Sequence[ScenarioStrategy]]): The strategies to execute. + Can be a list of ScenarioStrategy enum members. If None, uses the default aggregate from the scenario's configuration. dataset_config (Optional[DatasetConfiguration]): Configuration for the dataset source. Use this to specify dataset names or maximum dataset size from the CLI. diff --git a/pyrit/scenario/core/scenario_strategy.py b/pyrit/scenario/core/scenario_strategy.py index ccbc0f6c67..9ac5df43e5 100644 --- a/pyrit/scenario/core/scenario_strategy.py +++ b/pyrit/scenario/core/scenario_strategy.py @@ -205,7 +205,7 @@ def normalize_strategies(cls: type[T], strategies: set[T]) -> set[T]: print_deprecation_message( old_item="ScenarioStrategy.normalize_strategies", new_item="ScenarioStrategy.expand", - removed_in="0.8.0", + removed_in="0.15.0", ) return set(cls.expand(strategies)) @@ -275,151 +275,6 @@ def resolve(cls: type[T], strategies: Sequence[Any] | None, *, default: T) -> li result.append(item) return result - @classmethod - def prepare_scenario_strategies( - cls: type[T], - strategies: Sequence[T | ScenarioCompositeStrategy] | None = None, - *, - default_aggregate: T | None = None, - ) -> list[ScenarioCompositeStrategy]: - """ - Prepare and normalize scenario strategies for use in a scenario. - - This helper method simplifies scenario initialization by: - 1. Handling None input with sensible defaults - 2. Auto-wrapping bare ScenarioStrategy instances into ScenarioCompositeStrategy - 3. Expanding aggregate tags (like EASY, ALL) into concrete strategies - 4. Validating compositions according to the strategy's rules - - This eliminates boilerplate code in scenario __init__ methods. - - Args: - strategies (Sequence[T | ScenarioCompositeStrategy] | None): The strategies to prepare. - Can be a mix of bare strategy enums and composite strategies. - If None, uses default_aggregate to determine defaults. - If an empty sequence, returns an empty list (useful for baseline-only execution). - default_aggregate (T | None): The aggregate strategy to use when strategies is None. - Common values: MyStrategy.ALL, MyStrategy.EASY. If None when strategies is None, - raises ValueError. - - Returns: - List[ScenarioCompositeStrategy]: Normalized list of composite strategies ready for use. - May be empty if an empty sequence was explicitly provided. - - Raises: - ValueError: If strategies is None and default_aggregate is None, or if compositions - are invalid according to validate_composition(). - """ - print_deprecation_message( - old_item="ScenarioStrategy.prepare_scenario_strategies", - new_item="ScenarioStrategy.resolve and Scenario._prepare_strategies", - removed_in="0.8.0", - ) - # Handle None input with default aggregate - if strategies is None: - if default_aggregate is None: - raise ValueError( - f"Either strategies or default_aggregate must be provided. " - f"Common defaults: {cls.__name__}.ALL, {cls.__name__}.EASY" - ) - - # Expand the default aggregate into concrete strategies - expanded = cls.normalize_strategies({default_aggregate}) - # Wrap each in a ScenarioCompositeStrategy - composite_strategies = [ScenarioCompositeStrategy(strategies=[strategy]) for strategy in expanded] - else: - # Process the provided strategies - composite_strategies = [] - for item in strategies: - if isinstance(item, ScenarioCompositeStrategy): - # Already a composite, use as-is - composite_strategies.append(item) - elif isinstance(item, cls): - # Bare strategy enum - wrap it in a composite - composite_strategies.append(ScenarioCompositeStrategy(strategies=[item])) - else: - # Not our strategy type - skip or could raise error - # For now, skip to allow flexibility - pass - - # Allow empty list if explicitly provided (for baseline-only execution) - if not composite_strategies: - if strategies is not None and len(strategies) == 0: - return [] - raise ValueError( - f"No valid {cls.__name__} strategies provided. " - f"Provide at least one {cls.__name__} enum or ScenarioCompositeStrategy." - ) - - # Normalize compositions (expands aggregates, validates compositions) - return ScenarioCompositeStrategy.normalize_compositions(composite_strategies, strategy_type=cls) - - @classmethod - def supports_composition(cls: type[T]) -> bool: - """ - Indicate whether this strategy type supports composition. - - By default, strategies do NOT support composition (only single strategies allowed). - Subclasses that support composition (e.g., FoundryStrategy) should override this - to return True and implement validate_composition() to enforce their specific rules. - - Returns: - bool: True if composition is supported, False otherwise. - """ - print_deprecation_message( - old_item="ScenarioStrategy.supports_composition", - new_item="FoundryComposite dataclass", - removed_in="0.8.0", - ) - return False - - @classmethod - def validate_composition(cls: type[T], strategies: Sequence[T]) -> None: - """ - Validate whether the given strategies can be composed together. - - The base implementation checks supports_composition() and raises an error if - composition is not supported and multiple strategies are provided. - - Subclasses that support composition should override this method to define their - specific composition rules (e.g., "no more than one attack strategy"). - - Args: - strategies (Sequence[T]): The strategies to validate for composition. - - Raises: - ValueError: If the composition is invalid according to the subclass's rules. - The error message should clearly explain what rule was violated. - - Examples: - # EncodingStrategy doesn't support composition (uses default) - >>> EncodingStrategy.validate_composition([EncodingStrategy.Base64, EncodingStrategy.ROT13]) - ValueError: EncodingStrategy does not support composition. Each strategy must be used individually. - - # FoundryStrategy allows composition but with rules - >>> FoundryStrategy.validate_composition([FoundryStrategy.Crescendo, FoundryStrategy.MultiTurn]) - ValueError: Cannot compose multiple attack strategies: ['crescendo', 'multi_turn'] - """ - print_deprecation_message( - old_item="ScenarioStrategy.validate_composition", - new_item="FoundryComposite dataclass", - removed_in="0.8.0", - ) - if not strategies: - raise ValueError("Cannot validate empty strategy list") - - # Filter to only instances of this strategy type - typed_strategies = [s for s in strategies if isinstance(s, cls)] - - # Default rule: if composition is not supported, only single strategies allowed - if not cls.supports_composition() and len(typed_strategies) > 1: - raise ValueError( - f"{cls.__name__} does not support composition. " - f"Each strategy must be used individually. " - f"Received: {[s.value for s in typed_strategies]}" - ) - - class ScenarioCompositeStrategy: """ Represents a composition of one or more attack strategies. @@ -473,6 +328,12 @@ def __init__(self, *, strategies: Sequence[ScenarioStrategy]) -> None: if not strategies: raise ValueError("strategies list cannot be empty") + print_deprecation_message( + old_item="ScenarioCompositeStrategy", + new_item="FoundryComposite (from pyrit.scenario.scenarios.foundry)", + removed_in="0.15.0", + ) + self._strategies = list(strategies) if len(self._strategies) == 1: self._name = str(self._strategies[0].value) @@ -520,13 +381,6 @@ def extract_single_strategy_values( Raises: ValueError: If any composite contains multiple strategies. """ - from pyrit.common.deprecation import print_deprecation_message - - print_deprecation_message( - old_item="ScenarioCompositeStrategy.extract_single_strategy_values", - new_item="[s.value for s in self._scenario_strategies]", - removed_in="0.8.0", - ) # Check that all composites are single-strategy multi_strategy_composites = [comp for comp in composites if not comp.is_single_strategy] if multi_strategy_composites: @@ -576,8 +430,8 @@ def get_composite_name(strategies: Sequence[ScenarioStrategy]) -> str: print_deprecation_message( old_item="ScenarioCompositeStrategy.get_composite_name", - new_item="ScenarioCompositeStrategy.name property", - removed_in="0.8.0", + new_item="FoundryComposite.name", + removed_in="0.15.0", ) if not strategies: raise ValueError("Cannot generate name for empty strategy list") @@ -627,13 +481,6 @@ def normalize_compositions( # Error: Cannot mix aggregate with concrete in same composition [ScenarioCompositeStrategy(strategies=[EASY, Base64])] -> ValueError """ - from pyrit.common.deprecation import print_deprecation_message - - print_deprecation_message( - old_item="ScenarioCompositeStrategy.normalize_compositions", - new_item="ScenarioStrategy.resolve and Scenario._prepare_strategies", - removed_in="0.8.0", - ) if not compositions: raise ValueError("Compositions list cannot be empty") @@ -679,9 +526,7 @@ def normalize_compositions( ScenarioCompositeStrategy(strategies=[strategy]) for strategy in expanded ) else: - # Concrete composition - validate and preserve as-is - strategy_type.validate_composition(typed_strategies) - # Keep the composite (name is auto-generated from strategies) + # Concrete composition - preserve as-is (single-strategy composites are always valid) normalized_compositions.append(composite) if not normalized_compositions: diff --git a/pyrit/scenario/scenarios/foundry/__init__.py b/pyrit/scenario/scenarios/foundry/__init__.py index f8659040d5..2407b5674c 100644 --- a/pyrit/scenario/scenarios/foundry/__init__.py +++ b/pyrit/scenario/scenarios/foundry/__init__.py @@ -4,11 +4,13 @@ """Foundry scenario classes.""" from pyrit.scenario.scenarios.foundry.red_team_agent import ( + FoundryComposite, FoundryStrategy, RedTeamAgent, ) __all__ = [ + "FoundryComposite", "FoundryStrategy", "RedTeamAgent", ] diff --git a/pyrit/scenario/scenarios/foundry/red_team_agent.py b/pyrit/scenario/scenarios/foundry/red_team_agent.py index 2b9658b650..c33161ce12 100644 --- a/pyrit/scenario/scenarios/foundry/red_team_agent.py +++ b/pyrit/scenario/scenarios/foundry/red_team_agent.py @@ -17,7 +17,7 @@ from typing import TYPE_CHECKING, Any, Optional, TypeVar from pyrit.auth import get_azure_openai_auth -from pyrit.common import apply_defaults +from pyrit.common import REQUIRED_VALUE, apply_defaults from pyrit.datasets import TextJailBreak from pyrit.executor.attack import ( CrescendoAttack, @@ -60,13 +60,14 @@ from pyrit.prompt_normalizer.prompt_converter_configuration import ( PromptConverterConfiguration, ) +from pyrit.prompt_target import PromptTarget from pyrit.prompt_target.common.prompt_chat_target import PromptChatTarget from pyrit.prompt_target.openai.openai_chat_target import OpenAIChatTarget from pyrit.scenario.core.atomic_attack import AtomicAttack from pyrit.scenario.core.attack_technique import AttackTechnique from pyrit.scenario.core.dataset_configuration import DatasetConfiguration from pyrit.scenario.core.scenario import Scenario -from pyrit.scenario.core.scenario_strategy import ScenarioStrategy +from pyrit.scenario.core.scenario_strategy import ScenarioCompositeStrategy, ScenarioStrategy if TYPE_CHECKING: from pyrit.executor.attack.core.attack_strategy import AttackStrategy @@ -91,9 +92,11 @@ class FoundryComposite: @property def name(self) -> str: """Return a human-readable name for this composite.""" - attack_name = self.attack.value if self.attack else "baseline" if not self.converters: - return attack_name + return self.attack.value if self.attack else "baseline" + if self.attack is None and len(self.converters) == 1: + return self.converters[0].value + attack_name = self.attack.value if self.attack else "baseline" converter_names = ", ".join(c.value for c in self.converters) return f"ComposedStrategy({attack_name}, {converter_names})" @@ -267,23 +270,92 @@ def __init__( ) self._scenario_composites: list[FoundryComposite] = [] - def _prepare_strategies(self, strategies: Sequence[ScenarioStrategy] | None) -> list[ScenarioStrategy]: + @apply_defaults + async def initialize_async( + self, + *, + objective_target: PromptTarget = REQUIRED_VALUE, # type: ignore[assignment] + scenario_strategies: Optional[Sequence["FoundryStrategy | FoundryComposite | ScenarioCompositeStrategy"]] = None, + dataset_config: Optional[DatasetConfiguration] = None, + max_concurrency: int = 10, + max_retries: int = 0, + memory_labels: Optional[dict[str, str]] = None, + ) -> None: + """ + Initialize the scenario. + + Args: + objective_target (PromptTarget): The target system to attack. + scenario_strategies (Optional[Sequence[FoundryStrategy | FoundryComposite | ScenarioCompositeStrategy]]): The + strategies to execute. Accepts bare FoundryStrategy enum members, FoundryComposite + objects (for pairing an attack with converters), or a mix of both. Passing + ScenarioCompositeStrategy is deprecated — use FoundryComposite instead. + If None, uses the default aggregate (EASY). + dataset_config (Optional[DatasetConfiguration]): Configuration for the dataset source. + max_concurrency (int): Maximum number of concurrent attack executions. Defaults to 10. + max_retries (int): Maximum number of retries on failure. Defaults to 0. + memory_labels (Optional[dict[str, str]]): Labels to attach to all memory entries. + """ + await super().initialize_async( + objective_target=objective_target, + scenario_strategies=scenario_strategies, # type: ignore[arg-type] + dataset_config=dataset_config, + max_concurrency=max_concurrency, + max_retries=max_retries, + memory_labels=memory_labels, + ) + + def _prepare_strategies( # type: ignore[override] + self, + strategies: "Optional[Sequence[FoundryStrategy | FoundryComposite | ScenarioCompositeStrategy]]", + ) -> list[ScenarioStrategy]: """ Resolve strategies and build FoundryComposite objects. - Each resolved FoundryStrategy becomes a separate FoundryComposite. - Strategies tagged "attack" populate composite.attack; strategies tagged - "converter" populate composite.converters as a single-converter composite. + Accepts bare FoundryStrategy members (each becomes its own composite) or + FoundryComposite objects (used as-is, enabling attack+converter pairings). + None resolves to the default strategy aggregate. Args: - strategies: Optional sequence of FoundryStrategy members (or None to use default). + strategies: FoundryStrategy enums, FoundryComposite objects, or None for default. Returns: - list[ScenarioStrategy]: The resolved strategies stored in self._scenario_strategies. + list[ScenarioStrategy]: Flat list of constituent strategies for base-class tracking. """ - resolved = FoundryStrategy.resolve(strategies, default=self.get_default_strategy()) - self._scenario_composites = [self._strategy_to_composite(s) for s in resolved] - return list(resolved) + if strategies is None: + resolved = FoundryStrategy.resolve(None, default=self.get_default_strategy()) + self._scenario_composites = [self._strategy_to_composite(s) for s in resolved] + return list(resolved) + + # Process in input order, expanding aggregates for bare strategies in-place + composites: list[FoundryComposite] = [] + flat: list[ScenarioStrategy] = [] + seen: set[FoundryStrategy] = set() + + for item in strategies: + if isinstance(item, ScenarioCompositeStrategy): + # Legacy backward-compat: convert to FoundryComposite (ScenarioCompositeStrategy + # is deprecated — use FoundryComposite directly instead) + foundry_strats = [s for s in item.strategies if isinstance(s, FoundryStrategy)] + if foundry_strats: + item = FoundryComposite(attack=foundry_strats[0], converters=foundry_strats[1:]) + else: + continue + + if isinstance(item, FoundryComposite): + composites.append(item) + if item.attack: + flat.append(item.attack) + flat.extend(item.converters) + else: + for s in FoundryStrategy.resolve([item], default=self.get_default_strategy()): + if s not in seen: + seen.add(s) + composites.append(self._strategy_to_composite(s)) + flat.append(s) + + self._scenario_composites = composites + return flat @staticmethod def _strategy_to_composite(strategy: ScenarioStrategy) -> "FoundryComposite": diff --git a/tests/unit/scenario/test_foundry.py b/tests/unit/scenario/test_foundry.py index a920511c96..9c133a1ec4 100644 --- a/tests/unit/scenario/test_foundry.py +++ b/tests/unit/scenario/test_foundry.py @@ -15,8 +15,8 @@ from pyrit.prompt_converter import Base64Converter from pyrit.prompt_target import PromptTarget from pyrit.prompt_target.common.prompt_chat_target import PromptChatTarget -from pyrit.scenario import AtomicAttack, DatasetConfiguration -from pyrit.scenario.foundry import FoundryStrategy, RedTeamAgent +from pyrit.scenario import AtomicAttack, DatasetConfiguration, ScenarioCompositeStrategy +from pyrit.scenario.foundry import FoundryComposite, FoundryStrategy, RedTeamAgent from pyrit.score import FloatScaleThresholdScorer, TrueFalseScorer @@ -735,3 +735,98 @@ async def test_scenario_atomic_attack_count_matches_strategies( ) # Should have at least as many runs as specific strategies provided assert scenario.atomic_attack_count >= len(strategies) + + @patch.dict( + "os.environ", + { + "AZURE_OPENAI_GPT4O_UNSAFE_CHAT_ENDPOINT": "https://test.openai.azure.com/", + "AZURE_OPENAI_GPT4O_UNSAFE_CHAT_KEY": "test-key", + "AZURE_OPENAI_GPT4O_UNSAFE_CHAT_MODEL": "gpt-4", + }, + ) + @pytest.mark.asyncio + async def test_initialize_with_foundry_composite_directly( + self, mock_objective_target, mock_objective_scorer, mock_memory_seed_groups, mock_dataset_config + ): + """FoundryComposite objects passed to initialize_async are used as-is.""" + composite = FoundryComposite(attack=FoundryStrategy.Crescendo, converters=[FoundryStrategy.Base64]) + + with patch.object(RedTeamAgent, "_resolve_seed_groups", return_value=mock_memory_seed_groups): + scenario = RedTeamAgent( + attack_scoring_config=AttackScoringConfig(objective_scorer=mock_objective_scorer), + include_baseline=False, + ) + await scenario.initialize_async( + objective_target=mock_objective_target, + scenario_strategies=[composite], + dataset_config=mock_dataset_config, + ) + + assert len(scenario._scenario_composites) == 1 + result = scenario._scenario_composites[0] + assert result.attack == FoundryStrategy.Crescendo + assert result.converters == [FoundryStrategy.Base64] + assert result.name == "ComposedStrategy(crescendo, base64)" + + @patch.dict( + "os.environ", + { + "AZURE_OPENAI_GPT4O_UNSAFE_CHAT_ENDPOINT": "https://test.openai.azure.com/", + "AZURE_OPENAI_GPT4O_UNSAFE_CHAT_KEY": "test-key", + "AZURE_OPENAI_GPT4O_UNSAFE_CHAT_MODEL": "gpt-4", + }, + ) + @pytest.mark.asyncio + async def test_initialize_with_mixed_composites_and_strategies( + self, mock_objective_target, mock_objective_scorer, mock_memory_seed_groups, mock_dataset_config + ): + """A mix of bare FoundryStrategy and FoundryComposite can be passed together.""" + composite = FoundryComposite(attack=FoundryStrategy.Crescendo, converters=[FoundryStrategy.Base64]) + + with patch.object(RedTeamAgent, "_resolve_seed_groups", return_value=mock_memory_seed_groups): + scenario = RedTeamAgent( + attack_scoring_config=AttackScoringConfig(objective_scorer=mock_objective_scorer), + include_baseline=False, + ) + await scenario.initialize_async( + objective_target=mock_objective_target, + scenario_strategies=[composite, FoundryStrategy.ROT13], + dataset_config=mock_dataset_config, + ) + + assert len(scenario._scenario_composites) == 2 + assert scenario._scenario_composites[0].attack == FoundryStrategy.Crescendo + assert scenario._scenario_composites[1].attack is None + assert scenario._scenario_composites[1].converters == [FoundryStrategy.ROT13] + + @patch.dict( + "os.environ", + { + "AZURE_OPENAI_GPT4O_UNSAFE_CHAT_ENDPOINT": "https://test.openai.azure.com/", + "AZURE_OPENAI_GPT4O_UNSAFE_CHAT_KEY": "test-key", + "AZURE_OPENAI_GPT4O_UNSAFE_CHAT_MODEL": "gpt-4", + }, + ) + @pytest.mark.asyncio + async def test_initialize_converts_scenario_composite_strategy_to_foundry_composite( + self, mock_objective_target, mock_objective_scorer, mock_memory_seed_groups, mock_dataset_config + ): + """ScenarioCompositeStrategy passed to initialize_async is converted to FoundryComposite.""" + legacy = ScenarioCompositeStrategy(strategies=[FoundryStrategy.Crescendo, FoundryStrategy.Base64]) + + with patch.object(RedTeamAgent, "_resolve_seed_groups", return_value=mock_memory_seed_groups): + scenario = RedTeamAgent( + attack_scoring_config=AttackScoringConfig(objective_scorer=mock_objective_scorer), + include_baseline=False, + ) + await scenario.initialize_async( + objective_target=mock_objective_target, + scenario_strategies=[legacy], # type: ignore[arg-type] + dataset_config=mock_dataset_config, + ) + + assert len(scenario._scenario_composites) == 1 + result = scenario._scenario_composites[0] + assert result.attack == FoundryStrategy.Crescendo + assert result.converters == [FoundryStrategy.Base64] + diff --git a/tests/unit/scenario/test_strategy_validation.py b/tests/unit/scenario/test_strategy_validation.py index 6b8c457af9..bb8839d6c2 100644 --- a/tests/unit/scenario/test_strategy_validation.py +++ b/tests/unit/scenario/test_strategy_validation.py @@ -3,6 +3,8 @@ """Unit tests for strategy composition validation.""" +import warnings + import pytest from pyrit.scenario import ScenarioCompositeStrategy @@ -11,32 +13,13 @@ from pyrit.scenario.garak import EncodingStrategy -class TestStrategyValidation: - """Test validation of strategy compositions.""" - - def test_encoding_validation_allows_single_strategy(self): - """Test that encoding validation allows single strategies.""" - # Should not raise - EncodingStrategy.validate_composition([EncodingStrategy.Base64]) - - def test_encoding_validation_rejects_composition(self): - """Test that encoding validation rejects composed strategies.""" - with pytest.raises(ValueError, match="EncodingStrategy does not support composition"): - EncodingStrategy.validate_composition([EncodingStrategy.Base64, EncodingStrategy.ROT13]) - - def test_foundry_validation_allows_single_strategy(self): - """Test that foundry validation allows single strategies.""" - # Should not raise - FoundryStrategy.validate_composition([FoundryStrategy.Base64]) - - class TestFoundryComposite: """Tests for FoundryComposite dataclass construction and naming.""" def test_converter_only_composite_name(self): - """Test name for a composite with only a converter strategy.""" + """Test name for a composite with only a converter strategy — matches old single-strategy convention.""" composite = FoundryComposite(attack=None, converters=[FoundryStrategy.Base64]) - assert composite.name == "ComposedStrategy(baseline, base64)" + assert composite.name == "base64" def test_attack_only_composite_name(self): """Test name for a composite with only an attack strategy.""" @@ -61,6 +44,26 @@ def test_empty_composite_defaults(self): assert composite.converters == [] +class TestScenarioCompositeStrategyDeprecation: + """Test that ScenarioCompositeStrategy emits deprecation warnings.""" + + def test_init_emits_deprecation_warning(self): + """Creating a ScenarioCompositeStrategy should emit a DeprecationWarning.""" + with warnings.catch_warnings(record=True) as w: + warnings.simplefilter("always") + ScenarioCompositeStrategy(strategies=[EncodingStrategy.Base64]) + assert any(issubclass(warning.category, DeprecationWarning) for warning in w) + assert any("ScenarioCompositeStrategy" in str(warning.message) for warning in w) + + def test_init_warning_mentions_foundry_composite(self): + """The deprecation warning should point users to FoundryComposite.""" + with warnings.catch_warnings(record=True) as w: + warnings.simplefilter("always") + ScenarioCompositeStrategy(strategies=[EncodingStrategy.Base64]) + messages = [str(warning.message) for warning in w if issubclass(warning.category, DeprecationWarning)] + assert any("FoundryComposite" in msg for msg in messages) + + class TestScenarioCompositeStrategyExtraction: """Test extraction of strategy values from composite strategies.""" From b2e84e07dc458ba1daa350de84544a3c468e7ccc Mon Sep 17 00:00:00 2001 From: Richard Lundeen Date: Thu, 16 Apr 2026 17:13:11 -0700 Subject: [PATCH 3/8] cleanup --- doc/code/scenarios/0_scenarios.ipynb | 2 +- doc/code/scenarios/0_scenarios.py | 2 +- pyrit/scenario/core/scenario_strategy.py | 6 ++---- pyrit/scenario/scenarios/foundry/red_team_agent.py | 3 +++ 4 files changed, 7 insertions(+), 6 deletions(-) diff --git a/doc/code/scenarios/0_scenarios.ipynb b/doc/code/scenarios/0_scenarios.ipynb index 868cd01394..f460dde8b6 100644 --- a/doc/code/scenarios/0_scenarios.ipynb +++ b/doc/code/scenarios/0_scenarios.ipynb @@ -56,7 +56,7 @@ "1. **Strategy Enum**: Create a `ScenarioStrategy` enum that defines the available strategies for your scenario.\n", " - Each enum member is defined as `(value, tags)` where value is a string and tags is a set of strings\n", " - Include an `ALL` aggregate strategy that expands to all available strategies\n", - " - Optionally implement `supports_composition()` and `validate_composition()` for strategy composition rules\n", + " - Optionally override `_prepare_strategies()` for custom composition logic (see `FoundryComposite`)\n", "\n", "2. **Scenario Class**: Extend `Scenario` and implement these abstract methods:\n", " - `get_strategy_class()`: Return your strategy enum class\n", diff --git a/doc/code/scenarios/0_scenarios.py b/doc/code/scenarios/0_scenarios.py index 8335c7a248..ece568b90e 100644 --- a/doc/code/scenarios/0_scenarios.py +++ b/doc/code/scenarios/0_scenarios.py @@ -62,7 +62,7 @@ # 1. **Strategy Enum**: Create a `ScenarioStrategy` enum that defines the available strategies for your scenario. # - Each enum member is defined as `(value, tags)` where value is a string and tags is a set of strings # - Include an `ALL` aggregate strategy that expands to all available strategies -# - Optionally implement `supports_composition()` and `validate_composition()` for strategy composition rules +# - Optionally override `_prepare_strategies()` for custom composition logic (see `FoundryComposite`) # # 2. **Scenario Class**: Extend `Scenario` and implement these abstract methods: # - `get_strategy_class()`: Return your strategy enum class diff --git a/pyrit/scenario/core/scenario_strategy.py b/pyrit/scenario/core/scenario_strategy.py index 9ac5df43e5..325282b5e6 100644 --- a/pyrit/scenario/core/scenario_strategy.py +++ b/pyrit/scenario/core/scenario_strategy.py @@ -453,11 +453,9 @@ def normalize_compositions( Each aggregate expansion creates separate single-strategy compositions. Concrete strategy compositions are preserved together as single compositions. - This method also validates compositions according to the strategy's rules via validate_composition(). - Args: compositions (List[ScenarioCompositeStrategy]): List of composite strategies to normalize. - strategy_type (type[T]): The strategy enum type to use for normalization and validation. + strategy_type (type[T]): The strategy enum type to use for normalization. Returns: List[ScenarioCompositeStrategy]: Normalized list of composite strategies with aggregates expanded. @@ -465,7 +463,7 @@ def normalize_compositions( Raises: ValueError: If compositions is empty, contains empty compositions, mixes aggregates with concrete strategies in the same composition, - has multiple aggregates in one composition, or violates validate_composition() rules. + or has multiple aggregates in one composition. Example:: diff --git a/pyrit/scenario/scenarios/foundry/red_team_agent.py b/pyrit/scenario/scenarios/foundry/red_team_agent.py index c33161ce12..0df3bbe3ec 100644 --- a/pyrit/scenario/scenarios/foundry/red_team_agent.py +++ b/pyrit/scenario/scenarios/foundry/red_team_agent.py @@ -296,6 +296,9 @@ async def initialize_async( max_retries (int): Maximum number of retries on failure. Defaults to 0. memory_labels (Optional[dict[str, str]]): Labels to attach to all memory entries. """ + # This override exists purely for type-widening: FoundryComposite is a dataclass, + # not a ScenarioStrategy enum member, so the base class signature would reject it. + # All logic lives in _prepare_strategies (also overridden below). await super().initialize_async( objective_target=objective_target, scenario_strategies=scenario_strategies, # type: ignore[arg-type] From 0502182100fcac8929bde31cc8e8e0b7d619c8f4 Mon Sep 17 00:00:00 2001 From: Richard Lundeen Date: Thu, 16 Apr 2026 17:20:28 -0700 Subject: [PATCH 4/8] adding better docs --- doc/code/scenarios/0_scenarios.ipynb | 12 +++--------- doc/code/scenarios/0_scenarios.py | 12 +++--------- doc/code/scenarios/1_scenario_parameters.ipynb | 10 +++++----- doc/code/scenarios/1_scenario_parameters.py | 10 +++++----- pyrit/scenario/core/scenario_strategy.py | 6 ++++-- .../scenario/scenarios/foundry/red_team_agent.py | 15 +++++++++++++++ 6 files changed, 35 insertions(+), 30 deletions(-) diff --git a/doc/code/scenarios/0_scenarios.ipynb b/doc/code/scenarios/0_scenarios.ipynb index f460dde8b6..0be3bfe52d 100644 --- a/doc/code/scenarios/0_scenarios.ipynb +++ b/doc/code/scenarios/0_scenarios.ipynb @@ -108,7 +108,6 @@ " Scenario,\n", " ScenarioStrategy,\n", ")\n", - "from pyrit.scenario.core.scenario_strategy import ScenarioCompositeStrategy\n", "from pyrit.score.true_false.true_false_scorer import TrueFalseScorer\n", "from pyrit.setup import initialize_pyrit_async\n", "\n", @@ -162,19 +161,14 @@ " Build atomic attacks based on selected strategies.\n", "\n", " This method is called by initialize_async() after strategies are prepared.\n", - " Use self._scenario_composites to access the selected strategies.\n", + " Use self._scenario_strategies to access the resolved strategy list.\n", " \"\"\"\n", " atomic_attacks = []\n", "\n", " # objective_target is guaranteed to be non-None by parent class validation\n", " assert self._objective_target is not None\n", "\n", - " # Extract individual strategy values from the composites\n", - " selected_strategies = ScenarioCompositeStrategy.extract_single_strategy_values(\n", - " self._scenario_composites, strategy_type=MyStrategy\n", - " )\n", - "\n", - " for strategy in selected_strategies:\n", + " for strategy in self._scenario_strategies:\n", " # self._dataset_config is set by the parent class\n", " seed_groups = self._dataset_config.get_all_seed_groups()\n", "\n", @@ -185,7 +179,7 @@ " )\n", " atomic_attacks.append(\n", " AtomicAttack(\n", - " atomic_attack_name=strategy,\n", + " atomic_attack_name=strategy.value,\n", " attack=attack,\n", " seed_groups=seed_groups, # type: ignore[arg-type]\n", " memory_labels=self._memory_labels,\n", diff --git a/doc/code/scenarios/0_scenarios.py b/doc/code/scenarios/0_scenarios.py index ece568b90e..54ae36cb9e 100644 --- a/doc/code/scenarios/0_scenarios.py +++ b/doc/code/scenarios/0_scenarios.py @@ -96,7 +96,6 @@ Scenario, ScenarioStrategy, ) -from pyrit.scenario.core.scenario_strategy import ScenarioCompositeStrategy from pyrit.score.true_false.true_false_scorer import TrueFalseScorer from pyrit.setup import initialize_pyrit_async @@ -150,19 +149,14 @@ async def _get_atomic_attacks_async(self) -> list[AtomicAttack]: Build atomic attacks based on selected strategies. This method is called by initialize_async() after strategies are prepared. - Use self._scenario_composites to access the selected strategies. + Use self._scenario_strategies to access the resolved strategy list. """ atomic_attacks = [] # objective_target is guaranteed to be non-None by parent class validation assert self._objective_target is not None - # Extract individual strategy values from the composites - selected_strategies = ScenarioCompositeStrategy.extract_single_strategy_values( - self._scenario_composites, strategy_type=MyStrategy - ) - - for strategy in selected_strategies: + for strategy in self._scenario_strategies: # self._dataset_config is set by the parent class seed_groups = self._dataset_config.get_all_seed_groups() @@ -173,7 +167,7 @@ async def _get_atomic_attacks_async(self) -> list[AtomicAttack]: ) atomic_attacks.append( AtomicAttack( - atomic_attack_name=strategy, + atomic_attack_name=strategy.value, attack=attack, seed_groups=seed_groups, # type: ignore[arg-type] memory_labels=self._memory_labels, diff --git a/doc/code/scenarios/1_scenario_parameters.ipynb b/doc/code/scenarios/1_scenario_parameters.ipynb index 04d3fd6d55..5fddeb0b70 100644 --- a/doc/code/scenarios/1_scenario_parameters.ipynb +++ b/doc/code/scenarios/1_scenario_parameters.ipynb @@ -160,8 +160,8 @@ "id": "10", "metadata": {}, "source": [ - "**Composite strategies** — multiple converters applied together in sequence using\n", - "`ScenarioCompositeStrategy`:" + "**Composite strategies** — pair an attack with one or more converters using `FoundryComposite`.\n", + "For example, to run Crescendo with Base64 encoding applied:" ] }, { @@ -171,9 +171,9 @@ "metadata": {}, "outputs": [], "source": [ - "from pyrit.scenario import ScenarioCompositeStrategy\n", + "from pyrit.scenario.scenarios.foundry import FoundryComposite\n", "\n", - "composite_strategy = [ScenarioCompositeStrategy(strategies=[FoundryStrategy.Caesar, FoundryStrategy.CharSwap])]" + "composite_strategy = [FoundryComposite(attack=FoundryStrategy.Crescendo, converters=[FoundryStrategy.Base64])]" ] }, { @@ -194,7 +194,7 @@ "scenario_strategies = [\n", " FoundryStrategy.Base64,\n", " FoundryStrategy.Binary,\n", - " ScenarioCompositeStrategy(strategies=[FoundryStrategy.Caesar, FoundryStrategy.CharSwap]),\n", + " FoundryComposite(attack=FoundryStrategy.Crescendo, converters=[FoundryStrategy.Caesar]),\n", "]" ] }, diff --git a/doc/code/scenarios/1_scenario_parameters.py b/doc/code/scenarios/1_scenario_parameters.py index f62a9b85a0..6e9ee8ee81 100644 --- a/doc/code/scenarios/1_scenario_parameters.py +++ b/doc/code/scenarios/1_scenario_parameters.py @@ -79,13 +79,13 @@ aggregate_strategy = [FoundryStrategy.EASY] # %% [markdown] -# **Composite strategies** — multiple converters applied together in sequence using -# `ScenarioCompositeStrategy`: +# **Composite strategies** — pair an attack with one or more converters using `FoundryComposite`. +# For example, to run Crescendo with Base64 encoding applied: # %% -from pyrit.scenario import ScenarioCompositeStrategy +from pyrit.scenario.scenarios.foundry import FoundryComposite -composite_strategy = [ScenarioCompositeStrategy(strategies=[FoundryStrategy.Caesar, FoundryStrategy.CharSwap])] +composite_strategy = [FoundryComposite(attack=FoundryStrategy.Crescendo, converters=[FoundryStrategy.Base64])] # %% [markdown] # You can mix all three types in a single list: @@ -94,7 +94,7 @@ scenario_strategies = [ FoundryStrategy.Base64, FoundryStrategy.Binary, - ScenarioCompositeStrategy(strategies=[FoundryStrategy.Caesar, FoundryStrategy.CharSwap]), + FoundryComposite(attack=FoundryStrategy.Crescendo, converters=[FoundryStrategy.Caesar]), ] # %% [markdown] diff --git a/pyrit/scenario/core/scenario_strategy.py b/pyrit/scenario/core/scenario_strategy.py index 325282b5e6..cbbae6eb09 100644 --- a/pyrit/scenario/core/scenario_strategy.py +++ b/pyrit/scenario/core/scenario_strategy.py @@ -331,7 +331,8 @@ def __init__(self, *, strategies: Sequence[ScenarioStrategy]) -> None: print_deprecation_message( old_item="ScenarioCompositeStrategy", new_item="FoundryComposite (from pyrit.scenario.scenarios.foundry)", - removed_in="0.15.0", + # Extended to 0.18.0 to give external callers (e.g. Foundry) time to migrate. + removed_in="0.18.0", ) self._strategies = list(strategies) @@ -431,7 +432,8 @@ def get_composite_name(strategies: Sequence[ScenarioStrategy]) -> str: print_deprecation_message( old_item="ScenarioCompositeStrategy.get_composite_name", new_item="FoundryComposite.name", - removed_in="0.15.0", + # Extended to 0.18.0 to give external callers (e.g. Foundry) time to migrate. + removed_in="0.18.0", ) if not strategies: raise ValueError("Cannot generate name for empty strategy list") diff --git a/pyrit/scenario/scenarios/foundry/red_team_agent.py b/pyrit/scenario/scenarios/foundry/red_team_agent.py index 0df3bbe3ec..954282a04c 100644 --- a/pyrit/scenario/scenarios/foundry/red_team_agent.py +++ b/pyrit/scenario/scenarios/foundry/red_team_agent.py @@ -89,6 +89,21 @@ class FoundryComposite: attack: "FoundryStrategy | None" converters: "list[FoundryStrategy]" = field(default_factory=list) + def __post_init__(self) -> None: + if self.attack is not None and "attack" not in self.attack.tags: + raise ValueError( + f"FoundryComposite.attack must be an attack-tagged strategy " + f"(e.g., Crescendo, MultiTurn), got '{self.attack.value}'. " + f"Converter strategies belong in the converters list." + ) + attack_in_converters = [s for s in self.converters if "attack" in s.tags] + if attack_in_converters: + raise ValueError( + f"FoundryComposite.converters must only contain converter strategies, " + f"but got attack strategies: {[s.value for s in attack_in_converters]}. " + f"Pass an attack strategy via the attack parameter instead." + ) + @property def name(self) -> str: """Return a human-readable name for this composite.""" From 6ddef966874de36bbdb543b65d1ab31ed9232c52 Mon Sep 17 00:00:00 2001 From: Richard Lundeen Date: Thu, 16 Apr 2026 17:30:28 -0700 Subject: [PATCH 5/8] fixing warnings and pre-commit --- pyrit/scenario/core/scenario_strategy.py | 1 + .../scenarios/foundry/red_team_agent.py | 32 ++++++++--- tests/unit/scenario/test_foundry.py | 2 +- .../unit/scenario/test_strategy_validation.py | 54 +------------------ 4 files changed, 27 insertions(+), 62 deletions(-) diff --git a/pyrit/scenario/core/scenario_strategy.py b/pyrit/scenario/core/scenario_strategy.py index cbbae6eb09..7af549a8ec 100644 --- a/pyrit/scenario/core/scenario_strategy.py +++ b/pyrit/scenario/core/scenario_strategy.py @@ -275,6 +275,7 @@ def resolve(cls: type[T], strategies: Sequence[Any] | None, *, default: T) -> li result.append(item) return result + class ScenarioCompositeStrategy: """ Represents a composition of one or more attack strategies. diff --git a/pyrit/scenario/scenarios/foundry/red_team_agent.py b/pyrit/scenario/scenarios/foundry/red_team_agent.py index 954282a04c..63d84cec39 100644 --- a/pyrit/scenario/scenarios/foundry/red_team_agent.py +++ b/pyrit/scenario/scenarios/foundry/red_team_agent.py @@ -14,7 +14,7 @@ from collections.abc import Sequence from dataclasses import dataclass, field from inspect import signature -from typing import TYPE_CHECKING, Any, Optional, TypeVar +from typing import TYPE_CHECKING, Any, Optional, TypeVar, cast from pyrit.auth import get_azure_openai_auth from pyrit.common import REQUIRED_VALUE, apply_defaults @@ -90,6 +90,7 @@ class FoundryComposite: converters: "list[FoundryStrategy]" = field(default_factory=list) def __post_init__(self) -> None: + """Validate that attack and converter slots contain correctly tagged strategies.""" if self.attack is not None and "attack" not in self.attack.tags: raise ValueError( f"FoundryComposite.attack must be an attack-tagged strategy " @@ -110,7 +111,7 @@ def name(self) -> str: if not self.converters: return self.attack.value if self.attack else "baseline" if self.attack is None and len(self.converters) == 1: - return self.converters[0].value + return str(self.converters[0].value) attack_name = self.attack.value if self.attack else "baseline" converter_names = ", ".join(c.value for c in self.converters) return f"ComposedStrategy({attack_name}, {converter_names})" @@ -290,7 +291,9 @@ async def initialize_async( self, *, objective_target: PromptTarget = REQUIRED_VALUE, # type: ignore[assignment] - scenario_strategies: Optional[Sequence["FoundryStrategy | FoundryComposite | ScenarioCompositeStrategy"]] = None, + scenario_strategies: Optional[ + Sequence["FoundryStrategy | FoundryComposite | ScenarioCompositeStrategy"] + ] = None, dataset_config: Optional[DatasetConfiguration] = None, max_concurrency: int = 10, max_retries: int = 0, @@ -301,7 +304,8 @@ async def initialize_async( Args: objective_target (PromptTarget): The target system to attack. - scenario_strategies (Optional[Sequence[FoundryStrategy | FoundryComposite | ScenarioCompositeStrategy]]): The + scenario_strategies (Optional[Sequence[FoundryStrategy | FoundryComposite | + ScenarioCompositeStrategy]]): The strategies to execute. Accepts bare FoundryStrategy enum members, FoundryComposite objects (for pairing an attack with converters), or a mix of both. Passing ScenarioCompositeStrategy is deprecated — use FoundryComposite instead. @@ -316,7 +320,7 @@ async def initialize_async( # All logic lives in _prepare_strategies (also overridden below). await super().initialize_async( objective_target=objective_target, - scenario_strategies=scenario_strategies, # type: ignore[arg-type] + scenario_strategies=scenario_strategies, dataset_config=dataset_config, max_concurrency=max_concurrency, max_retries=max_retries, @@ -341,7 +345,7 @@ def _prepare_strategies( # type: ignore[override] list[ScenarioStrategy]: Flat list of constituent strategies for base-class tracking. """ if strategies is None: - resolved = FoundryStrategy.resolve(None, default=self.get_default_strategy()) + resolved = FoundryStrategy.resolve(None, default=cast(FoundryStrategy, self.get_default_strategy())) self._scenario_composites = [self._strategy_to_composite(s) for s in resolved] return list(resolved) @@ -366,7 +370,7 @@ def _prepare_strategies( # type: ignore[override] flat.append(item.attack) flat.extend(item.converters) else: - for s in FoundryStrategy.resolve([item], default=self.get_default_strategy()): + for s in FoundryStrategy.resolve([item], default=cast(FoundryStrategy, self.get_default_strategy())): if s not in seen: seen.add(s) composites.append(self._strategy_to_composite(s)) @@ -377,7 +381,16 @@ def _prepare_strategies( # type: ignore[override] @staticmethod def _strategy_to_composite(strategy: ScenarioStrategy) -> "FoundryComposite": - """Wrap a single FoundryStrategy in a FoundryComposite.""" + """ + Wrap a single FoundryStrategy in a FoundryComposite. + + Returns: + FoundryComposite: Attack-slotted composite for attack-tagged strategies; + converter-slotted composite otherwise. + + Raises: + ValueError: If strategy is not a FoundryStrategy instance. + """ if not isinstance(strategy, FoundryStrategy): raise ValueError(f"Expected FoundryStrategy, got {type(strategy)}") if "attack" in strategy.tags: @@ -424,6 +437,9 @@ def _get_attack_from_strategy(self, composite: FoundryComposite) -> AtomicAttack Returns: AtomicAttack: The configured atomic attack. + + Raises: + ValueError: If a converter strategy in the composite is not recognized. """ attack: AttackStrategy[Any, Any] diff --git a/tests/unit/scenario/test_foundry.py b/tests/unit/scenario/test_foundry.py index 9c133a1ec4..cccd9e4c73 100644 --- a/tests/unit/scenario/test_foundry.py +++ b/tests/unit/scenario/test_foundry.py @@ -807,6 +807,7 @@ async def test_initialize_with_mixed_composites_and_strategies( "AZURE_OPENAI_GPT4O_UNSAFE_CHAT_MODEL": "gpt-4", }, ) + @pytest.mark.filterwarnings("ignore::DeprecationWarning") @pytest.mark.asyncio async def test_initialize_converts_scenario_composite_strategy_to_foundry_composite( self, mock_objective_target, mock_objective_scorer, mock_memory_seed_groups, mock_dataset_config @@ -829,4 +830,3 @@ async def test_initialize_converts_scenario_composite_strategy_to_foundry_compos result = scenario._scenario_composites[0] assert result.attack == FoundryStrategy.Crescendo assert result.converters == [FoundryStrategy.Base64] - diff --git a/tests/unit/scenario/test_strategy_validation.py b/tests/unit/scenario/test_strategy_validation.py index bb8839d6c2..0e6fa92de4 100644 --- a/tests/unit/scenario/test_strategy_validation.py +++ b/tests/unit/scenario/test_strategy_validation.py @@ -5,8 +5,6 @@ import warnings -import pytest - from pyrit.scenario import ScenarioCompositeStrategy from pyrit.scenario.foundry import FoundryStrategy from pyrit.scenario.foundry.red_team_agent import FoundryComposite @@ -40,7 +38,7 @@ def test_attack_with_multiple_converters_composite_name(self): def test_empty_composite_defaults(self): """Test that FoundryComposite defaults converters to empty list.""" - composite = FoundryComposite(attack=FoundryStrategy.Base64) + composite = FoundryComposite(attack=FoundryStrategy.Crescendo) assert composite.converters == [] @@ -62,53 +60,3 @@ def test_init_warning_mentions_foundry_composite(self): ScenarioCompositeStrategy(strategies=[EncodingStrategy.Base64]) messages = [str(warning.message) for warning in w if issubclass(warning.category, DeprecationWarning)] assert any("FoundryComposite" in msg for msg in messages) - - -class TestScenarioCompositeStrategyExtraction: - """Test extraction of strategy values from composite strategies.""" - - def test_extract_single_strategy_values_with_single_strategies(self): - """Test extracting values from single-strategy composites.""" - composites = [ - ScenarioCompositeStrategy(strategies=[EncodingStrategy.Base64]), - ScenarioCompositeStrategy(strategies=[EncodingStrategy.ROT13]), - ScenarioCompositeStrategy(strategies=[EncodingStrategy.Atbash]), - ] - - values = ScenarioCompositeStrategy.extract_single_strategy_values(composites, strategy_type=EncodingStrategy) - - assert values == {"base64", "rot13", "atbash"} - - def test_extract_single_strategy_values_filters_by_type(self): - """Test that extraction filters by strategy type.""" - composites = [ - ScenarioCompositeStrategy(strategies=[EncodingStrategy.Base64]), - ScenarioCompositeStrategy(strategies=[FoundryStrategy.ROT13]), - ] - - # Extract only EncodingStrategy values - encoding_values = ScenarioCompositeStrategy.extract_single_strategy_values( - composites, strategy_type=EncodingStrategy - ) - assert encoding_values == {"base64"} - - # Extract only FoundryStrategy values - foundry_values = ScenarioCompositeStrategy.extract_single_strategy_values( - composites, strategy_type=FoundryStrategy - ) - assert foundry_values == {"rot13"} - - def test_extract_single_strategy_values_rejects_multi_strategy_composites(self): - """Test that extraction raises error if any composite has multiple strategies.""" - composites = [ - ScenarioCompositeStrategy(strategies=[FoundryStrategy.Base64]), - ScenarioCompositeStrategy(strategies=[FoundryStrategy.ROT13, FoundryStrategy.Atbash]), # Multi-strategy! - ] - - with pytest.raises(ValueError, match="extract_single_strategy_values.*requires all composites"): - ScenarioCompositeStrategy.extract_single_strategy_values(composites, strategy_type=FoundryStrategy) - - def test_extract_single_strategy_values_with_empty_list(self): - """Test that extraction handles empty composite list.""" - values = ScenarioCompositeStrategy.extract_single_strategy_values([], strategy_type=EncodingStrategy) - assert values == set() From c6c8851e4ab0b0658a8b2607ebfbcfef5eeb07b2 Mon Sep 17 00:00:00 2001 From: Richard Lundeen Date: Thu, 16 Apr 2026 18:00:40 -0700 Subject: [PATCH 6/8] fixing pre-commit --- .../scenarios/foundry/red_team_agent.py | 26 +++++--- tests/unit/scenario/test_foundry.py | 62 +++++++++++++++++++ 2 files changed, 79 insertions(+), 9 deletions(-) diff --git a/pyrit/scenario/scenarios/foundry/red_team_agent.py b/pyrit/scenario/scenarios/foundry/red_team_agent.py index 63d84cec39..dd09f6798f 100644 --- a/pyrit/scenario/scenarios/foundry/red_team_agent.py +++ b/pyrit/scenario/scenarios/foundry/red_team_agent.py @@ -90,7 +90,13 @@ class FoundryComposite: converters: "list[FoundryStrategy]" = field(default_factory=list) def __post_init__(self) -> None: - """Validate that attack and converter slots contain correctly tagged strategies.""" + """ + Validate that attack and converter slots contain correctly tagged strategies. + + Raises: + ValueError: If attack slot contains a non-attack-tagged strategy, or if + converters list contains an attack-tagged strategy. + """ if self.attack is not None and "attack" not in self.attack.tags: raise ValueError( f"FoundryComposite.attack must be an attack-tagged strategy " @@ -304,8 +310,7 @@ async def initialize_async( Args: objective_target (PromptTarget): The target system to attack. - scenario_strategies (Optional[Sequence[FoundryStrategy | FoundryComposite | - ScenarioCompositeStrategy]]): The + scenario_strategies (Sequence[FoundryStrategy | FoundryComposite | ScenarioCompositeStrategy] | None): The strategies to execute. Accepts bare FoundryStrategy enum members, FoundryComposite objects (for pairing an attack with converters), or a mix of both. Passing ScenarioCompositeStrategy is deprecated — use FoundryComposite instead. @@ -345,7 +350,7 @@ def _prepare_strategies( # type: ignore[override] list[ScenarioStrategy]: Flat list of constituent strategies for base-class tracking. """ if strategies is None: - resolved = FoundryStrategy.resolve(None, default=cast(FoundryStrategy, self.get_default_strategy())) + resolved = FoundryStrategy.resolve(None, default=cast("FoundryStrategy", self.get_default_strategy())) self._scenario_composites = [self._strategy_to_composite(s) for s in resolved] return list(resolved) @@ -357,12 +362,15 @@ def _prepare_strategies( # type: ignore[override] for item in strategies: if isinstance(item, ScenarioCompositeStrategy): # Legacy backward-compat: convert to FoundryComposite (ScenarioCompositeStrategy - # is deprecated — use FoundryComposite directly instead) + # is deprecated — use FoundryComposite directly instead). + # Route by tags rather than position: the first attack-tagged strategy + # becomes `attack`; all converter-tagged strategies become `converters`. foundry_strats = [s for s in item.strategies if isinstance(s, FoundryStrategy)] - if foundry_strats: - item = FoundryComposite(attack=foundry_strats[0], converters=foundry_strats[1:]) - else: + if not foundry_strats: continue + attack_strat = next((s for s in foundry_strats if "attack" in s.tags), None) + converter_strats = [s for s in foundry_strats if "attack" not in s.tags] + item = FoundryComposite(attack=attack_strat, converters=converter_strats) if isinstance(item, FoundryComposite): composites.append(item) @@ -370,7 +378,7 @@ def _prepare_strategies( # type: ignore[override] flat.append(item.attack) flat.extend(item.converters) else: - for s in FoundryStrategy.resolve([item], default=cast(FoundryStrategy, self.get_default_strategy())): + for s in FoundryStrategy.resolve([item], default=cast("FoundryStrategy", self.get_default_strategy())): if s not in seen: seen.add(s) composites.append(self._strategy_to_composite(s)) diff --git a/tests/unit/scenario/test_foundry.py b/tests/unit/scenario/test_foundry.py index cccd9e4c73..c4fdc7579f 100644 --- a/tests/unit/scenario/test_foundry.py +++ b/tests/unit/scenario/test_foundry.py @@ -830,3 +830,65 @@ async def test_initialize_converts_scenario_composite_strategy_to_foundry_compos result = scenario._scenario_composites[0] assert result.attack == FoundryStrategy.Crescendo assert result.converters == [FoundryStrategy.Base64] + + @patch.dict( + "os.environ", + { + "AZURE_OPENAI_GPT4O_UNSAFE_CHAT_ENDPOINT": "https://test.openai.azure.com/", + "AZURE_OPENAI_GPT4O_UNSAFE_CHAT_KEY": "test-key", + "AZURE_OPENAI_GPT4O_UNSAFE_CHAT_MODEL": "gpt-4", + }, + ) + @pytest.mark.filterwarnings("ignore::DeprecationWarning") + @pytest.mark.asyncio + async def test_initialize_converts_converter_first_composite_strategy( + self, mock_objective_target, mock_objective_scorer, mock_memory_seed_groups, mock_dataset_config + ): + """Converter-first ScenarioCompositeStrategy is routed by tags, not position.""" + legacy = ScenarioCompositeStrategy(strategies=[FoundryStrategy.Base64, FoundryStrategy.Crescendo]) + + with patch.object(RedTeamAgent, "_resolve_seed_groups", return_value=mock_memory_seed_groups): + scenario = RedTeamAgent( + attack_scoring_config=AttackScoringConfig(objective_scorer=mock_objective_scorer), + include_baseline=False, + ) + await scenario.initialize_async( + objective_target=mock_objective_target, + scenario_strategies=[legacy], # type: ignore[arg-type] + dataset_config=mock_dataset_config, + ) + + result = scenario._scenario_composites[0] + assert result.attack == FoundryStrategy.Crescendo + assert result.converters == [FoundryStrategy.Base64] + + @patch.dict( + "os.environ", + { + "AZURE_OPENAI_GPT4O_UNSAFE_CHAT_ENDPOINT": "https://test.openai.azure.com/", + "AZURE_OPENAI_GPT4O_UNSAFE_CHAT_KEY": "test-key", + "AZURE_OPENAI_GPT4O_UNSAFE_CHAT_MODEL": "gpt-4", + }, + ) + @pytest.mark.filterwarnings("ignore::DeprecationWarning") + @pytest.mark.asyncio + async def test_initialize_converts_converter_only_composite_strategy( + self, mock_objective_target, mock_objective_scorer, mock_memory_seed_groups, mock_dataset_config + ): + """Converter-only ScenarioCompositeStrategy maps to attack=None.""" + legacy = ScenarioCompositeStrategy(strategies=[FoundryStrategy.Base64, FoundryStrategy.ROT13]) + + with patch.object(RedTeamAgent, "_resolve_seed_groups", return_value=mock_memory_seed_groups): + scenario = RedTeamAgent( + attack_scoring_config=AttackScoringConfig(objective_scorer=mock_objective_scorer), + include_baseline=False, + ) + await scenario.initialize_async( + objective_target=mock_objective_target, + scenario_strategies=[legacy], # type: ignore[arg-type] + dataset_config=mock_dataset_config, + ) + + result = scenario._scenario_composites[0] + assert result.attack is None + assert set(result.converters) == {FoundryStrategy.Base64, FoundryStrategy.ROT13} From 9f6a1e1091b531526ec8112cb116d9751b9c59f6 Mon Sep 17 00:00:00 2001 From: Richard Lundeen Date: Fri, 17 Apr 2026 13:12:25 -0700 Subject: [PATCH 7/8] pr feedback --- doc/scanner/foundry.ipynb | 12 ++++++------ doc/scanner/foundry.py | 12 ++++++------ .../scenarios/foundry/red_team_agent.py | 12 ++++++------ tests/unit/scenario/test_strategy_validation.py | 17 +++++++++++++++++ 4 files changed, 35 insertions(+), 18 deletions(-) diff --git a/doc/scanner/foundry.ipynb b/doc/scanner/foundry.ipynb index 999d7317bb..dbceb7fe0d 100644 --- a/doc/scanner/foundry.ipynb +++ b/doc/scanner/foundry.ipynb @@ -213,13 +213,13 @@ "source": [ "## Strategy Composition\n", "\n", - "You can combine multiple converters into a single composite strategy using\n", - "`ScenarioCompositeStrategy`. Each converter in the composite is applied in sequence.\n", + "You can pair a multi-turn attack with one or more converter strategies using `FoundryComposite`.\n", + "Each converter in the composite is applied in sequence before the attack runs.\n", "\n", "```python\n", - "from pyrit.scenario import ScenarioCompositeStrategy\n", + "from pyrit.scenario.scenarios.foundry import FoundryComposite\n", "\n", - "composed = ScenarioCompositeStrategy(strategies=[FoundryStrategy.Caesar, FoundryStrategy.CharSwap])\n", + "composed = FoundryComposite(attack=FoundryStrategy.Crescendo, converters=[FoundryStrategy.Caesar, FoundryStrategy.CharSwap])\n", "```" ] }, @@ -230,8 +230,8 @@ "metadata": {}, "outputs": [], "source": [ - "# from pyrit.scenario import ScenarioCompositeStrategy\n", - "# composed = ScenarioCompositeStrategy(strategies=[FoundryStrategy.Caesar, FoundryStrategy.CharSwap])\n", + "# from pyrit.scenario.scenarios.foundry import FoundryComposite\n", + "# composed = FoundryComposite(attack=FoundryStrategy.Crescendo, converters=[FoundryStrategy.Caesar, FoundryStrategy.CharSwap])\n", "# scenario_strategies = [FoundryStrategy.Base64, composed]" ] }, diff --git a/doc/scanner/foundry.py b/doc/scanner/foundry.py index f5545181ac..2b71b3f5b8 100644 --- a/doc/scanner/foundry.py +++ b/doc/scanner/foundry.py @@ -75,18 +75,18 @@ # %% [markdown] # ## Strategy Composition # -# You can combine multiple converters into a single composite strategy using -# `ScenarioCompositeStrategy`. Each converter in the composite is applied in sequence. +# You can pair a multi-turn attack with one or more converter strategies using `FoundryComposite`. +# Each converter in the composite is applied in sequence before the attack runs. # # ```python -# from pyrit.scenario import ScenarioCompositeStrategy +# from pyrit.scenario.scenarios.foundry import FoundryComposite # -# composed = ScenarioCompositeStrategy(strategies=[FoundryStrategy.Caesar, FoundryStrategy.CharSwap]) +# composed = FoundryComposite(attack=FoundryStrategy.Crescendo, converters=[FoundryStrategy.Caesar, FoundryStrategy.CharSwap]) # ``` # %% -# from pyrit.scenario import ScenarioCompositeStrategy -# composed = ScenarioCompositeStrategy(strategies=[FoundryStrategy.Caesar, FoundryStrategy.CharSwap]) +# from pyrit.scenario.scenarios.foundry import FoundryComposite +# composed = FoundryComposite(attack=FoundryStrategy.Crescendo, converters=[FoundryStrategy.Caesar, FoundryStrategy.CharSwap]) # scenario_strategies = [FoundryStrategy.Base64, composed] # %% [markdown] diff --git a/pyrit/scenario/scenarios/foundry/red_team_agent.py b/pyrit/scenario/scenarios/foundry/red_team_agent.py index dd09f6798f..9f4878ca1b 100644 --- a/pyrit/scenario/scenarios/foundry/red_team_agent.py +++ b/pyrit/scenario/scenarios/foundry/red_team_agent.py @@ -95,7 +95,7 @@ def __post_init__(self) -> None: Raises: ValueError: If attack slot contains a non-attack-tagged strategy, or if - converters list contains an attack-tagged strategy. + converters list contains any non-converter-tagged strategy (including aggregates). """ if self.attack is not None and "attack" not in self.attack.tags: raise ValueError( @@ -103,12 +103,12 @@ def __post_init__(self) -> None: f"(e.g., Crescendo, MultiTurn), got '{self.attack.value}'. " f"Converter strategies belong in the converters list." ) - attack_in_converters = [s for s in self.converters if "attack" in s.tags] - if attack_in_converters: + misrouted = [s for s in self.converters if "converter" not in s.tags] + if misrouted: raise ValueError( - f"FoundryComposite.converters must only contain converter strategies, " - f"but got attack strategies: {[s.value for s in attack_in_converters]}. " - f"Pass an attack strategy via the attack parameter instead." + f"FoundryComposite.converters must only contain converter-tagged strategies, " + f"got {[s.value for s in misrouted]}. " + f"Attack strategies belong in the attack parameter; aggregates must be expanded first." ) @property diff --git a/tests/unit/scenario/test_strategy_validation.py b/tests/unit/scenario/test_strategy_validation.py index 0e6fa92de4..529028ad9c 100644 --- a/tests/unit/scenario/test_strategy_validation.py +++ b/tests/unit/scenario/test_strategy_validation.py @@ -3,6 +3,8 @@ """Unit tests for strategy composition validation.""" +import pytest + import warnings from pyrit.scenario import ScenarioCompositeStrategy @@ -41,6 +43,21 @@ def test_empty_composite_defaults(self): composite = FoundryComposite(attack=FoundryStrategy.Crescendo) assert composite.converters == [] + def test_converter_in_attack_slot_raises(self): + """Putting a converter-tagged strategy in the attack slot should raise.""" + with pytest.raises(ValueError, match="attack must be an attack-tagged strategy"): + FoundryComposite(attack=FoundryStrategy.Base64) + + def test_attack_in_converters_raises(self): + """Putting an attack-tagged strategy in converters should raise.""" + with pytest.raises(ValueError, match="converters must only contain converter-tagged"): + FoundryComposite(attack=None, converters=[FoundryStrategy.Crescendo]) + + def test_aggregate_in_converters_raises(self): + """Aggregates (e.g. EASY) in converters slot should fail early rather than silently later.""" + with pytest.raises(ValueError, match="converters must only contain converter-tagged"): + FoundryComposite(attack=None, converters=[FoundryStrategy.EASY]) + class TestScenarioCompositeStrategyDeprecation: """Test that ScenarioCompositeStrategy emits deprecation warnings.""" From 68b5dc526a5bcaceb870e20195d813c0d71d811c Mon Sep 17 00:00:00 2001 From: Richard Lundeen Date: Fri, 17 Apr 2026 13:51:17 -0700 Subject: [PATCH 8/8] pr feedback --- doc/code/scenarios/0_scenarios.ipynb | 4 +- doc/code/scenarios/0_scenarios.py | 4 +- .../scenarios/1_scenario_parameters.ipynb | 7 +- doc/code/scenarios/1_scenario_parameters.py | 7 +- pyrit/scenario/core/scenario.py | 6 +- pyrit/scenario/core/scenario_strategy.py | 185 +----------------- .../scenarios/foundry/red_team_agent.py | 6 +- tests/unit/scenario/test_scenario.py | 31 ++- .../unit/scenario/test_strategy_validation.py | 4 +- 9 files changed, 45 insertions(+), 209 deletions(-) diff --git a/doc/code/scenarios/0_scenarios.ipynb b/doc/code/scenarios/0_scenarios.ipynb index 0be3bfe52d..8bf1c1a83c 100644 --- a/doc/code/scenarios/0_scenarios.ipynb +++ b/doc/code/scenarios/0_scenarios.ipynb @@ -405,8 +405,8 @@ "each objective directly to the target without any converters or multi-turn techniques. This is\n", "controlled by the `include_default_baseline` parameter (default: `True` for most scenarios).\n", "\n", - "To run *only* the baseline (no attack strategies), pass `scenario_strategies=[]` programmatically.\n", - "This is useful for establishing a refusal rate before applying attacks. See\n", + "To run *only* the baseline (no attack strategies), create a `RedTeamAgent` with\n", + "`include_baseline=True` (the default) and pass `scenario_strategies=None`. See\n", "[Scenario Parameters](./1_scenario_parameters.ipynb) for a working example." ] }, diff --git a/doc/code/scenarios/0_scenarios.py b/doc/code/scenarios/0_scenarios.py index 54ae36cb9e..4df620c808 100644 --- a/doc/code/scenarios/0_scenarios.py +++ b/doc/code/scenarios/0_scenarios.py @@ -193,8 +193,8 @@ async def _get_atomic_attacks_async(self) -> list[AtomicAttack]: # each objective directly to the target without any converters or multi-turn techniques. This is # controlled by the `include_default_baseline` parameter (default: `True` for most scenarios). # -# To run *only* the baseline (no attack strategies), pass `scenario_strategies=[]` programmatically. -# This is useful for establishing a refusal rate before applying attacks. See +# To run *only* the baseline (no attack strategies), create a `RedTeamAgent` with +# `include_baseline=True` (the default) and pass `scenario_strategies=None`. See # [Scenario Parameters](./1_scenario_parameters.ipynb) for a working example. # %% [markdown] diff --git a/doc/code/scenarios/1_scenario_parameters.ipynb b/doc/code/scenarios/1_scenario_parameters.ipynb index 5fddeb0b70..840ba82708 100644 --- a/doc/code/scenarios/1_scenario_parameters.ipynb +++ b/doc/code/scenarios/1_scenario_parameters.ipynb @@ -205,8 +205,9 @@ "source": [ "## Baseline Execution\n", "\n", - "Pass an empty `scenario_strategies` list to run a baseline-only scenario. The baseline sends each\n", - "objective directly to the target without any converters or multi-turn strategies. This is useful for:\n", + "The baseline sends each objective directly to the target without any converters or multi-turn\n", + "strategies. It is included automatically when `include_baseline=True` (the default). This is\n", + "useful for:\n", "\n", "- **Measuring default defenses** — how does the target respond to unmodified harmful prompts?\n", "- **Establishing comparison points** — compare baseline refusal rates against attack-enhanced runs\n", @@ -304,7 +305,7 @@ "baseline_scenario = RedTeamAgent()\n", "await baseline_scenario.initialize_async( # type: ignore\n", " objective_target=objective_target,\n", - " scenario_strategies=[], # Empty list = baseline only\n", + " scenario_strategies=None, # Uses default strategies; baseline is prepended automatically\n", " dataset_config=dataset_config,\n", ")\n", "baseline_result = await baseline_scenario.run_async() # type: ignore\n", diff --git a/doc/code/scenarios/1_scenario_parameters.py b/doc/code/scenarios/1_scenario_parameters.py index 6e9ee8ee81..c5df2d7297 100644 --- a/doc/code/scenarios/1_scenario_parameters.py +++ b/doc/code/scenarios/1_scenario_parameters.py @@ -100,8 +100,9 @@ # %% [markdown] # ## Baseline Execution # -# Pass an empty `scenario_strategies` list to run a baseline-only scenario. The baseline sends each -# objective directly to the target without any converters or multi-turn strategies. This is useful for: +# The baseline sends each objective directly to the target without any converters or multi-turn +# strategies. It is included automatically when `include_baseline=True` (the default). This is +# useful for: # # - **Measuring default defenses** — how does the target respond to unmodified harmful prompts? # - **Establishing comparison points** — compare baseline refusal rates against attack-enhanced runs @@ -111,7 +112,7 @@ baseline_scenario = RedTeamAgent() await baseline_scenario.initialize_async( # type: ignore objective_target=objective_target, - scenario_strategies=[], # Empty list = baseline only + scenario_strategies=None, # Uses default strategies; baseline is prepended automatically dataset_config=dataset_config, ) baseline_result = await baseline_scenario.run_async() # type: ignore diff --git a/pyrit/scenario/core/scenario.py b/pyrit/scenario/core/scenario.py index f53f675016..7b1e5338c1 100644 --- a/pyrit/scenario/core/scenario.py +++ b/pyrit/scenario/core/scenario.py @@ -191,14 +191,14 @@ def _prepare_strategies( Resolve strategy inputs into a concrete list for this scenario. The default implementation calls resolve() on the strategy class, which handles - None (use default), empty list (baseline-only), and aggregate expansion. + None (use default), empty list (also use default), and aggregate expansion. Subclasses with complex composition semantics (e.g., RedTeamAgent with FoundryComposite) should override this to build their own composite types. Args: - strategies: Strategy inputs from initialize_async. None means use default, - [] means baseline-only, otherwise a list of strategies to resolve. + strategies: Strategy inputs from initialize_async. None or [] both mean use + default; otherwise a list of strategies to resolve. Returns: list[ScenarioStrategy]: Ordered, deduplicated concrete strategies. diff --git a/pyrit/scenario/core/scenario_strategy.py b/pyrit/scenario/core/scenario_strategy.py index 7af549a8ec..ffa9d29825 100644 --- a/pyrit/scenario/core/scenario_strategy.py +++ b/pyrit/scenario/core/scenario_strategy.py @@ -248,14 +248,14 @@ def resolve(cls: type[T], strategies: Sequence[Any] | None, *, default: T) -> li backward compatibility. Args: - strategies (Sequence[Any] | None): Strategies to resolve. If None, expands the - default. If an empty list, returns empty (baseline-only execution). - default (T): Default aggregate strategy to use when strategies is None. + strategies (Sequence[Any] | None): Strategies to resolve. If None or empty, + expands the default. + default (T): Default aggregate strategy to use when strategies is None or empty. Returns: list[T]: Ordered, deduplicated list of concrete strategies. """ - if strategies is None: + if not strategies: return cls.expand({default}) result: list[T] = [] @@ -358,183 +358,6 @@ def is_single_strategy(self) -> bool: """Check if this composition contains only a single strategy.""" return len(self._strategies) == 1 - @staticmethod - def extract_single_strategy_values( - composites: Sequence[ScenarioCompositeStrategy], *, strategy_type: type[T] - ) -> set[str]: - """ - Extract strategy values from single-strategy composites. - - This is a helper method for scenarios that don't support composition and need - to filter or map strategies by their values. It flattens the composites into - a simple set of strategy values. - - This method enforces that all composites contain only a single strategy. If any - composite contains multiple strategies, a ValueError is raised. - - Args: - composites (Sequence[ScenarioCompositeStrategy]): List of composite strategies. - Each composite must contain only a single strategy. - strategy_type (type[T]): The strategy enum type to filter by. - - Returns: - Set[str]: Set of strategy values (e.g., {"base64", "rot13", "morse_code"}). - - Raises: - ValueError: If any composite contains multiple strategies. - """ - # Check that all composites are single-strategy - multi_strategy_composites = [comp for comp in composites if not comp.is_single_strategy] - if multi_strategy_composites: - composite_names = [comp.name for comp in multi_strategy_composites] - raise ValueError( - f"extract_single_strategy_values() requires all composites to contain a single strategy. " - f"Found composites with multiple strategies: {composite_names}" - ) - - return { - strategy.value - for composite in composites - for strategy in composite.strategies - if isinstance(strategy, strategy_type) - } - - @staticmethod - def get_composite_name(strategies: Sequence[ScenarioStrategy]) -> str: - """ - Generate a descriptive name for a composition of strategies. - - For single strategies, returns the strategy's value. - For multiple strategies, generates a name like "ComposedStrategy(base64, rot13)". - - Args: - strategies (Sequence[ScenarioStrategy]): The strategies to generate a name for. - - Returns: - str: The generated composite name. - - Raises: - ValueError: If strategies is empty. - - Example: - >>> # Single strategy - >>> name = ScenarioCompositeStrategy.get_composite_name([FoundryStrategy.Base64]) - >>> # Returns: "base64" - >>> - >>> # Multiple strategies - >>> name = ScenarioCompositeStrategy.get_composite_name([ - ... FoundryStrategy.Base64, - ... FoundryStrategy.Atbash - ... ]) - >>> # Returns: "ComposedStrategy(base64, atbash)" - """ - from pyrit.common.deprecation import print_deprecation_message - - print_deprecation_message( - old_item="ScenarioCompositeStrategy.get_composite_name", - new_item="FoundryComposite.name", - # Extended to 0.18.0 to give external callers (e.g. Foundry) time to migrate. - removed_in="0.18.0", - ) - if not strategies: - raise ValueError("Cannot generate name for empty strategy list") - - if len(strategies) == 1: - return str(strategies[0].value) - - strategy_names = ", ".join(s.value for s in strategies) - return f"ComposedStrategy({strategy_names})" - - @staticmethod - def normalize_compositions( - compositions: list[ScenarioCompositeStrategy], *, strategy_type: type[T] - ) -> list[ScenarioCompositeStrategy]: - """ - Normalize strategy compositions by expanding aggregates while preserving concrete compositions. - - Aggregate strategies are expanded into their constituent individual strategies. - Each aggregate expansion creates separate single-strategy compositions. - Concrete strategy compositions are preserved together as single compositions. - - Args: - compositions (List[ScenarioCompositeStrategy]): List of composite strategies to normalize. - strategy_type (type[T]): The strategy enum type to use for normalization. - - Returns: - List[ScenarioCompositeStrategy]: Normalized list of composite strategies with aggregates expanded. - - Raises: - ValueError: If compositions is empty, contains empty compositions, - mixes aggregates with concrete strategies in the same composition, - or has multiple aggregates in one composition. - - Example:: - - # Aggregate expands to individual strategies - [ScenarioCompositeStrategy(strategies=[EASY])] - -> [ScenarioCompositeStrategy(strategies=[Base64]), - ScenarioCompositeStrategy(strategies=[ROT13]), ...] - - # Concrete composition preserved - [ScenarioCompositeStrategy(strategies=[Base64, Atbash])] - -> [ScenarioCompositeStrategy(strategies=[Base64, Atbash])] - - # Error: Cannot mix aggregate with concrete in same composition - [ScenarioCompositeStrategy(strategies=[EASY, Base64])] -> ValueError - """ - if not compositions: - raise ValueError("Compositions list cannot be empty") - - aggregate_tags = strategy_type.get_aggregate_tags() - normalized_compositions: list[ScenarioCompositeStrategy] = [] - - for composite in compositions: - if not composite.strategies: - raise ValueError("Empty compositions are not allowed") - - # Filter to only strategies of the specified type - typed_strategies = [s for s in composite.strategies if isinstance(s, strategy_type)] - if not typed_strategies: - # No strategies of this type - skip - continue - - # Check if composition contains any aggregates - aggregates_in_composition = [s for s in typed_strategies if s.value in aggregate_tags] - concretes_in_composition = [s for s in typed_strategies if s.value not in aggregate_tags] - - # Error if mixing aggregates with concrete strategies - if aggregates_in_composition and concretes_in_composition: - raise ValueError( - f"Cannot mix aggregate strategies {[s.value for s in aggregates_in_composition]} " - f"with concrete strategies {[s.value for s in concretes_in_composition]} " - f"in the same composition. Aggregates must be in their own composition to be expanded." - ) - - # Error if multiple aggregates in same composition - if len(aggregates_in_composition) > 1: - raise ValueError( - f"Cannot compose multiple aggregate strategies together: " - f"{[s.value for s in aggregates_in_composition]}. " - f"Each aggregate must be in its own composition." - ) - - # If composition has an aggregate, expand it into individual strategies - if aggregates_in_composition: - aggregate = aggregates_in_composition[0] - expanded = strategy_type.normalize_strategies({aggregate}) - # Each expanded strategy becomes its own composition - normalized_compositions.extend( - ScenarioCompositeStrategy(strategies=[strategy]) for strategy in expanded - ) - else: - # Concrete composition - preserve as-is (single-strategy composites are always valid) - normalized_compositions.append(composite) - - if not normalized_compositions: - raise ValueError("No valid strategy compositions after normalization") - - return normalized_compositions - def __repr__(self) -> str: """ Get string representation of the composite strategy. diff --git a/pyrit/scenario/scenarios/foundry/red_team_agent.py b/pyrit/scenario/scenarios/foundry/red_team_agent.py index 9f4878ca1b..bf9332771d 100644 --- a/pyrit/scenario/scenarios/foundry/red_team_agent.py +++ b/pyrit/scenario/scenarios/foundry/red_team_agent.py @@ -341,15 +341,15 @@ def _prepare_strategies( # type: ignore[override] Accepts bare FoundryStrategy members (each becomes its own composite) or FoundryComposite objects (used as-is, enabling attack+converter pairings). - None resolves to the default strategy aggregate. + None and [] both resolve to the default strategy aggregate. Args: - strategies: FoundryStrategy enums, FoundryComposite objects, or None for default. + strategies: FoundryStrategy enums, FoundryComposite objects, or None/[] for default. Returns: list[ScenarioStrategy]: Flat list of constituent strategies for base-class tracking. """ - if strategies is None: + if not strategies: resolved = FoundryStrategy.resolve(None, default=cast("FoundryStrategy", self.get_default_strategy())) self._scenario_composites = [self._strategy_to_composite(s) for s in resolved] return list(resolved) diff --git a/tests/unit/scenario/test_scenario.py b/tests/unit/scenario/test_scenario.py index 7f02982015..d1a8505c81 100644 --- a/tests/unit/scenario/test_scenario.py +++ b/tests/unit/scenario/test_scenario.py @@ -688,7 +688,7 @@ class TestScenarioBaselineOnlyExecution: @pytest.mark.asyncio async def test_initialize_async_with_empty_strategies_and_baseline(self, mock_objective_target): - """Test that baseline-only execution works when include_baseline=True and strategies is empty.""" + """Test that baseline is included when include_baseline=True, regardless of strategies.""" from pyrit.models import SeedAttackGroup, SeedObjective # Create a scenario with include_default_baseline=True and TrueFalseScorer @@ -705,10 +705,10 @@ async def test_initialize_async_with_empty_strategies_and_baseline(self, mock_ob SeedAttackGroup(seeds=[SeedObjective(value="test objective 2")]), ] - # Initialize with empty strategies + # Initialize with None (default strategy) — [] also works, both expand defaults await scenario.initialize_async( objective_target=mock_objective_target, - scenario_strategies=[], # Empty list - baseline only + scenario_strategies=None, dataset_config=mock_dataset_config, ) @@ -734,10 +734,10 @@ async def test_baseline_only_execution_runs_successfully(self, mock_objective_ta SeedAttackGroup(seeds=[SeedObjective(value="test objective 1")]), ] - # Initialize with empty strategies + # Initialize with None — [] also expands defaults now, both are equivalent await scenario.initialize_async( objective_target=mock_objective_target, - scenario_strategies=[], # Empty list - baseline only + scenario_strategies=None, # same as [] now dataset_config=mock_dataset_config, ) @@ -754,7 +754,7 @@ async def test_baseline_only_execution_runs_successfully(self, mock_objective_ta @pytest.mark.asyncio async def test_empty_strategies_without_baseline_allows_initialization(self, mock_objective_target): - """Test that empty strategies without include_baseline allows initialization but fails at run time.""" + """Test that no strategies + no baseline allows initialization but fails at run time.""" scenario = ConcreteScenario( name="No Baseline Test", version=1, @@ -763,11 +763,10 @@ async def test_empty_strategies_without_baseline_allows_initialization(self, moc mock_dataset_config = MagicMock(spec=DatasetConfiguration) - # Empty strategies are now always allowed during initialization - # (no allow_empty parameter required) + # None strategies with no baseline: _get_atomic_attacks_async returns [] await scenario.initialize_async( objective_target=mock_objective_target, - scenario_strategies=[], # Empty list without baseline + scenario_strategies=None, dataset_config=mock_dataset_config, ) @@ -798,7 +797,7 @@ async def test_standalone_baseline_uses_dataset_config_seeds(self, mock_objectiv await scenario.initialize_async( objective_target=mock_objective_target, - scenario_strategies=[], + scenario_strategies=None, dataset_config=mock_dataset_config, ) @@ -807,6 +806,18 @@ async def test_standalone_baseline_uses_dataset_config_seeds(self, mock_objectiv assert baseline_attack.atomic_attack_name == "baseline" assert baseline_attack.seed_groups == expected_seeds + def test_empty_list_strategies_expands_defaults_same_as_none(self): + """Test that [] and None both expand to the default strategy set.""" + scenario = ConcreteScenario(name="Test", version=1) + strategy_class = scenario.get_strategy_class() + default = scenario.get_default_strategy() + + resolved_none = strategy_class.resolve(None, default=default) + resolved_empty = strategy_class.resolve([], default=default) + + assert resolved_none == resolved_empty + assert len(resolved_none) > 0 + class TestGetDefaultObjectiveScorer: """Tests for Scenario._get_default_objective_scorer method.""" diff --git a/tests/unit/scenario/test_strategy_validation.py b/tests/unit/scenario/test_strategy_validation.py index 529028ad9c..f4f0b03f21 100644 --- a/tests/unit/scenario/test_strategy_validation.py +++ b/tests/unit/scenario/test_strategy_validation.py @@ -3,10 +3,10 @@ """Unit tests for strategy composition validation.""" -import pytest - import warnings +import pytest + from pyrit.scenario import ScenarioCompositeStrategy from pyrit.scenario.foundry import FoundryStrategy from pyrit.scenario.foundry.red_team_agent import FoundryComposite