diff --git a/doc/code/scenarios/0_scenarios.ipynb b/doc/code/scenarios/0_scenarios.ipynb index 868cd0139..0be3bfe52 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", @@ -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 8335c7a24..54ae36cb9 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 @@ -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 04d3fd6d5..5fddeb0b7 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 f62a9b85a..6e9ee8ee8 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/doc/scanner/foundry.ipynb b/doc/scanner/foundry.ipynb index 999d7317b..dbceb7fe0 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 f5545181a..2b71b3f5b 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/core/dataset_configuration.py b/pyrit/scenario/core/dataset_configuration.py index 2b78ced2f..25cd9162c 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 7e91c53bd..f53f67501 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: @@ -114,8 +111,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,12 +183,34 @@ 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]], + ) -> 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, *, 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, @@ -210,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. @@ -246,11 +263,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 0252f6841..7af549a8e 100644 --- a/pyrit/scenario/core/scenario_strategy.py +++ b/pyrit/scenario/core/scenario_strategy.py @@ -202,157 +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.15.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)) - - return normalized_strategies + concrete.update(cls.get_strategies_by_tag(aggregate_tag)) + return [s for s in cls if s in concrete] @classmethod - def prepare_scenario_strategies( - cls: type[T], - strategies: Sequence[T | ScenarioCompositeStrategy] | None = None, - *, - default_aggregate: T | None = None, - ) -> list[ScenarioCompositeStrategy]: + def resolve(cls: type[T], strategies: Sequence[Any] | None, *, default: T) -> list[T]: """ - 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 + Resolve strategy inputs into a concrete, ordered, deduplicated list. - This eliminates boilerplate code in scenario __init__ methods. + 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[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. + 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[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(). + list[T]: Ordered, deduplicated list of concrete strategies. """ - # 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. + return cls.expand({default}) - Returns: - bool: True if composition is supported, False otherwise. - """ - 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'] - """ - 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]}" - ) + 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 class ScenarioCompositeStrategy: @@ -408,8 +329,19 @@ 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)", + # Extended to 0.18.0 to give external callers (e.g. Foundry) time to migrate. + removed_in="0.18.0", + ) + 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: @@ -496,6 +428,14 @@ 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="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") @@ -516,11 +456,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. @@ -528,7 +466,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:: @@ -589,9 +527,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/airt/content_harms.py b/pyrit/scenario/scenarios/airt/content_harms.py index d22ece85f..400e19667 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 fe53b62bb..57b6a3edf 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 9667cca82..8dfe49021 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 43354690c..41877e9ef 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 2caf2d6f7..9d201baf9 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 e2db719f4..e1d70457a 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/__init__.py b/pyrit/scenario/scenarios/foundry/__init__.py index f8659040d..2407b5674 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 2095827e0..9f4878ca1 100644 --- a/pyrit/scenario/scenarios/foundry/red_team_agent.py +++ b/pyrit/scenario/scenarios/foundry/red_team_agent.py @@ -12,11 +12,12 @@ 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 +from typing import TYPE_CHECKING, Any, Optional, TypeVar, cast 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, @@ -59,16 +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 ( - ScenarioCompositeStrategy, - ScenarioStrategy, -) +from pyrit.scenario.core.scenario_strategy import ScenarioCompositeStrategy, ScenarioStrategy if TYPE_CHECKING: from pyrit.executor.attack.core.attack_strategy import AttackStrategy @@ -77,6 +76,53 @@ 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) + + def __post_init__(self) -> None: + """ + 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 any non-converter-tagged strategy (including aggregates). + """ + 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." + ) + misrouted = [s for s in self.converters if "converter" not in s.tags] + if misrouted: + raise ValueError( + 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 + def name(self) -> str: + """Return a human-readable name for this composite.""" + if not self.converters: + return self.attack.value if self.attack else "baseline" + if self.attack is None and len(self.converters) == 1: + 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})" + + class FoundryStrategy(ScenarioStrategy): """ Strategies for attacks with tag-based categorization. @@ -154,47 +200,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 +290,120 @@ def __init__( include_default_baseline=include_baseline, scenario_result_id=scenario_result_id, ) + self._scenario_composites: list[FoundryComposite] = [] + + @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 (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. + 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. + """ + # 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, + 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. + + 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: FoundryStrategy enums, FoundryComposite objects, or None for default. + + Returns: + 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())) + 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). + # 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 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) + if item.attack: + flat.append(item.attack) + flat.extend(item.converters) + else: + 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)) + flat.append(s) + + self._scenario_composites = composites + return flat + + @staticmethod + def _strategy_to_composite(strategy: ScenarioStrategy) -> "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: + return FoundryComposite(attack=strategy) + return FoundryComposite(attack=None, converters=[strategy]) def _resolve_seed_groups(self) -> list[SeedAttackGroup]: """ @@ -316,49 +435,37 @@ 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). + ValueError: If a converter strategy in the composite is not recognized. """ 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 +515,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 1864c2a93..ccf455045 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 1e177e15b..b05a18ff6 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 0a1fae4b8..e1b5c6872 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 79a802d2e..1271bdaa8 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_foundry.py b/tests/unit/scenario/test_foundry.py index a920511c9..c4fdc7579 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,160 @@ 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.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 + ): + """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] + + @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} diff --git a/tests/unit/scenario/test_strategy_validation.py b/tests/unit/scenario/test_strategy_validation.py index 2721b9ac6..529028ad9 100644 --- a/tests/unit/scenario/test_strategy_validation.py +++ b/tests/unit/scenario/test_strategy_validation.py @@ -5,99 +5,75 @@ import pytest +import warnings + 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 -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]) - - 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] - ) - - 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] - ) - - -class TestScenarioCompositeStrategyExtraction: - """Test extraction of strategy values from composite strategies.""" +class TestFoundryComposite: + """Tests for FoundryComposite dataclass construction and naming.""" - 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]), - ] + def test_converter_only_composite_name(self): + """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 == "base64" - values = ScenarioCompositeStrategy.extract_single_strategy_values(composites, strategy_type=EncodingStrategy) + 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" - 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"} + 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)" - # Extract only FoundryStrategy values - foundry_values = ScenarioCompositeStrategy.extract_single_strategy_values( - composites, strategy_type=FoundryStrategy + 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 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() + assert composite.name == "ComposedStrategy(crescendo, base64, atbash)" + + def test_empty_composite_defaults(self): + """Test that FoundryComposite defaults converters to empty list.""" + 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.""" + + 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)