From 46949b5a50cfd27244a8b4ad1ebe320670c94c4e Mon Sep 17 00:00:00 2001 From: Richard Lundeen Date: Thu, 2 Jul 2026 19:10:54 -0700 Subject: [PATCH] MAINT: Delete legacy Scenario scaffolding (Phase D) Make _build_atomic_attacks_async(context) the sole abstract extension point on Scenario and delete the dead base machinery every scenario had to override around: the _get_atomic_attacks_async bridge, _get_attack_technique_factories, _build_display_group, _prepare_strategies, and the 0.16.0 baseline-injection deprecation shims. Add extra_factories to the matrix builder helpers so local factories (Leakage) stay on the one-line construction path. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../instructions/scenarios.instructions.md | 112 +++++---- doc/code/scenarios/0_scenarios.ipynb | 32 +-- doc/code/scenarios/0_scenarios.py | 30 +-- doc/scanner/1_pyrit_scan.ipynb | 7 +- doc/scanner/1_pyrit_scan.py | 7 +- .../registry/components/scenario_registry.py | 4 +- pyrit/scenario/core/attack_technique.py | 7 +- .../scenario/core/attack_technique_factory.py | 6 +- pyrit/scenario/core/dataset_configuration.py | 7 +- .../core/matrix_atomic_attack_builder.py | 22 +- pyrit/scenario/core/scenario.py | 226 +++--------------- pyrit/scenario/scenarios/adaptive/__init__.py | 6 +- .../scenarios/adaptive/adaptive_scenario.py | 17 +- .../scenarios/adaptive/selectors/__init__.py | 9 +- .../scenarios/adaptive/text_adaptive.py | 13 +- pyrit/scenario/scenarios/airt/cyber.py | 18 +- pyrit/scenario/scenarios/airt/jailbreak.py | 28 +-- pyrit/scenario/scenarios/airt/leakage.py | 40 ++-- pyrit/scenario/scenarios/airt/psychosocial.py | 27 +-- .../scenario/scenarios/airt/rapid_response.py | 2 +- pyrit/scenario/scenarios/airt/scam.py | 35 +-- .../scenarios/benchmark/adversarial.py | 12 +- pyrit/scenario/scenarios/foundry/__init__.py | 6 +- .../scenarios/foundry/red_team_agent.py | 48 +--- pyrit/scenario/scenarios/garak/__init__.py | 5 +- pyrit/scenario/scenarios/garak/doctor.py | 9 +- pyrit/scenario/scenarios/garak/encoding.py | 28 +-- .../scenario/scenarios/garak/web_injection.py | 78 +++--- tests/unit/scenario/airt/test_cyber.py | 23 +- tests/unit/scenario/airt/test_jailbreak.py | 20 +- tests/unit/scenario/airt/test_leakage.py | 4 +- tests/unit/scenario/airt/test_psychosocial.py | 11 +- .../unit/scenario/airt/test_rapid_response.py | 37 +-- tests/unit/scenario/airt/test_scam.py | 20 +- .../scenario/benchmark/test_adversarial.py | 62 ++--- .../core/test_attack_technique_factory.py | 6 +- .../core/test_baseline_deprecation.py | 215 ----------------- .../core/test_matrix_atomic_attack_builder.py | 28 +++ tests/unit/scenario/core/test_scenario.py | 133 +++-------- .../scenario/core/test_scenario_parameters.py | 5 +- .../core/test_scenario_partial_results.py | 5 +- .../unit/scenario/core/test_scenario_retry.py | 5 +- .../scenario/foundry/test_red_team_agent.py | 4 +- tests/unit/scenario/garak/test_doctor.py | 2 +- tests/unit/scenario/garak/test_encoding.py | 8 +- .../scenarios/adaptive/test_text_adaptive.py | 34 ++- 46 files changed, 452 insertions(+), 1011 deletions(-) delete mode 100644 tests/unit/scenario/core/test_baseline_deprecation.py diff --git a/.github/instructions/scenarios.instructions.md b/.github/instructions/scenarios.instructions.md index bd7a2bc7b7..ca7fae917d 100644 --- a/.github/instructions/scenarios.instructions.md +++ b/.github/instructions/scenarios.instructions.md @@ -39,9 +39,9 @@ For scenarios whose strategy enum is built dynamically (RapidResponse pattern), strategy class in a module-level `@cache`-decorated function and pass the result through the constructor — no classmethod indirection required. -4. **Optionally override `_get_atomic_attacks_async()`** — the base class provides a default - that uses the factory/registry pattern (see "AtomicAttack Construction" below). - Only override if your scenario needs custom attack construction logic. +4. **Implement `_build_atomic_attacks_async(self, *, context)`** — this is the single + abstract extension point every scenario must define (see "AtomicAttack Construction" below). + Matrix-shaped scenarios delegate to `build_matrix_atomic_attacks(context=...)` in one line. ## Constructor Pattern @@ -58,7 +58,7 @@ def __init__( if not objective_scorer: objective_scorer = self._get_default_scorer() - # 2. Store config objects for _get_atomic_attacks_async + # 2. Store config objects for _build_atomic_attacks_async self._scorer_config = AttackScoringConfig(objective_scorer=objective_scorer) # 3. Call super().__init__ — required args: version, strategy_class, objective_scorer @@ -80,7 +80,7 @@ Requirements: may opt into a one-release grace period via the class attribute `_brick_legacy_init = True`, which downgrades the error to a `DeprecationWarning(removed_in="0.16.0")`. The opt-out is removed in 0.16.0. -- **All constructor parameters must be optional** (default to `None`) so the registry can instantiate the scenario with no arguments for metadata introspection. Defer required-input validation to `initialize_async()` or `_get_atomic_attacks_async()`. `ScenarioRegistry._build_metadata` raises `TypeError` if `scenario_class()` cannot be called with no arguments. +- **All constructor parameters must be optional** (default to `None`) so the registry can instantiate the scenario with no arguments for metadata introspection. Defer required-input validation to `initialize_async()` or `_build_atomic_attacks_async()`. `ScenarioRegistry._build_metadata` raises `TypeError` if `scenario_class()` cannot be called with no arguments. - `super().__init__()` called with `version`, `strategy_class`, `default_strategy`, `default_dataset_config`, `objective_scorer` - complex objects like `adversarial_chat` or `objective_scorer` should be passed into the constructor. @@ -134,39 +134,70 @@ class MyStrategy(ScenarioStrategy): - Each member: `NAME = ("string_value", {tag_set})` - Aggregates expand to all strategies matching their tag -### `_build_display_group()` — Result Grouping +### Result grouping (`display_group`) -Override `_build_display_group()` on the `Scenario` base class to control how attack results are grouped for display: +`display_group` controls how attack results are aggregated for display. It is set per +`AtomicAttack` at construction time — there is no `_build_display_group` hook. When you build +via `build_matrix_atomic_attacks`/`MatrixAtomicAttackBuilder`, pass a `display_group_fn` +callback that maps each `MatrixCombo` to a group string: ```python -def _build_display_group(self, *, technique_name: str, seed_group_name: str) -> str: - # Default: group by technique name (most common) - return technique_name - - # Override examples: - # Group by dataset/harm category: return seed_group_name - # Cross-product: return f"{technique_name}_{seed_group_name}" +build_matrix_atomic_attacks( + context=context, + objective_scorer=self._objective_scorer, + display_group_fn=lambda combo: combo.technique_name, # default: group by technique + # Group by dataset/harm category: lambda combo: combo.dataset_name + # Cross-product: lambda combo: f"{combo.technique_name}_{combo.dataset_name}" +) ``` Note: `atomic_attack_name` must remain unique per `AtomicAttack` for correct resume behaviour. `display_group` controls user-facing aggregation only. -## AtomicAttack Construction — Default Base Class Behaviour +## AtomicAttack Construction — `_build_atomic_attacks_async(context)` + +Every scenario implements the single abstract extension point: + +```python +async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list[AtomicAttack]: + ... +``` + +`initialize_async` resolves the run's inputs once (objective target, strategies, dataset +config, memory labels, baseline flag, and seed groups), snapshots them into an immutable +`ScenarioContext`, calls this method, and then inserts the baseline centrally. Scenario authors +never read half-initialized `self._*` state to build attacks — read everything from `context`. -The `Scenario` base class provides a default `_get_atomic_attacks_async()` that uses the -factory/registry pattern. Scenarios that register their techniques via `_get_attack_technique_factories()` -get atomic-attack construction **for free** — no override needed. +### Zero-boilerplate matrix scenarios -The default implementation: -1. Calls `self._get_attack_technique_factories()` to get name→factory mapping - (defaults to reading every `AttackTechniqueFactory` registered in the - `AttackTechniqueRegistry` singleton) -2. Iterates over every (technique × dataset) pair from `self._dataset_config` -3. Calls `factory.create()` with `objective_target` and conditional scorer override - (also forwards any per-technique converters from `self._strategy_converters`, populated - from the CLI `--strategies :converter.` modifier, as `extra_request_converters`) -4. Uses `self._build_display_group()` for user-facing grouping -5. Builds `AtomicAttack` with unique `atomic_attack_name` = `"{technique}_{dataset}"` +Scenarios whose construction is the plain technique × dataset cross-product delegate to the +`build_matrix_atomic_attacks` helper in one line (see `Cyber`, `RapidResponse`): + +```python +from pyrit.scenario.core.matrix_atomic_attack_builder import build_matrix_atomic_attacks + +async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list[AtomicAttack]: + return build_matrix_atomic_attacks( + context=context, + objective_scorer=self._objective_scorer, + strategy_converters=self._strategy_converters, # optional CLI converter stacks + ) +``` + +`build_matrix_atomic_attacks`: +1. Calls `resolve_technique_factories(context=context)` to map the selected strategies to their + registered `AttackTechniqueFactory` instances (reads the `AttackTechniqueRegistry` singleton; + strategies with no registered factory are dropped). +2. Iterates every (technique × dataset) pair from `context.seed_groups_by_dataset`. +3. Calls `factory.create()` with the objective target, conditional scorer override, and any + per-technique converters (from `--strategies :converter.`) as + `extra_request_converters`. +4. Builds each `AtomicAttack` with a unique `atomic_attack_name` and a `display_group` + (customizable via `display_group_fn`). + +Scenarios needing extra axes (adversarial targets, caching, converter stacks) call +`MatrixAtomicAttackBuilder` directly; scenarios whose construction is composite or +per-objective build the `AtomicAttack` list themselves (see "Manual AtomicAttack construction"). ### AttackTechniqueFactory @@ -213,29 +244,28 @@ by the registry. Tests that exercise scenarios should reset both `AttackTechniqu and `TargetRegistry` and re-register a mock `adversarial_chat` so the catalog builder resolves without falling back to `OpenAIChatTarget`. -### Customization hooks (no need to override `_get_atomic_attacks_async`): -- **`_get_attack_technique_factories()`** — override to add/remove/replace factories -- **`_build_display_group()`** — override to change grouping (default: by technique) - -### When to override `_get_atomic_attacks_async`: -Only override when the scenario **cannot** use the factory/registry pattern — e.g., scenarios -with custom composite logic, per-strategy converter stacks, or non-standard attack construction. +### Baseline -Overrides that want baseline support must emit it themselves by calling `self._build_baseline_atomic_attack(seed_groups=...)` with the same seeds used for the strategy attacks and prepending the result. The base implementation emits baseline automatically; passing freshly resolved seeds reintroduces ADO 9012 (baseline-vs-strategy population divergence under `max_dataset_size`). +The baseline (a `PromptSendingAttack` over the run's seeds) is inserted **centrally** by +`Scenario.initialize_async` according to the scenario's `BASELINE_ATTACK_POLICY` class var and +the runtime `include_baseline` flag. `_build_atomic_attacks_async` must **never** prepend its own +baseline — doing so double-emits it and reintroduces baseline-vs-strategy population divergence +under `max_dataset_size`. -### Manual AtomicAttack construction (for overrides): +### Manual AtomicAttack construction: ```python AtomicAttack( - atomic_attack_name=strategy_name, # groups related attacks + atomic_attack_name=strategy_name, # must be unique per AtomicAttack attack=attack_instance, # AttackStrategy implementation seed_groups=list(seed_groups), # must be non-empty - memory_labels=self._memory_labels, # from base class + memory_labels=context.memory_labels, # from the context snapshot ) ``` - `seed_groups` must be non-empty — validate before constructing -- `self._objective_target` is only available after `initialize_async()` — don't access in `__init__` +- Read runtime inputs from `context`, not `self._*` — `self._objective_target` and + `self._scenario_strategies` are only populated after `initialize_async()` - Pass `memory_labels` to every AtomicAttack ## Exports @@ -248,4 +278,4 @@ New scenarios must be registered in `pyrit/scenario/__init__.py` as virtual pack - Forgetting `@apply_defaults` on `__init__` - Empty `seed_groups` passed to `AtomicAttack` - Missing `VERSION` class constant -- Missing `_async` suffix on `_get_atomic_attacks_async` +- Missing `_async` suffix on `_build_atomic_attacks_async` diff --git a/doc/code/scenarios/0_scenarios.ipynb b/doc/code/scenarios/0_scenarios.ipynb index 716deb78aa..757974420b 100644 --- a/doc/code/scenarios/0_scenarios.ipynb +++ b/doc/code/scenarios/0_scenarios.ipynb @@ -57,13 +57,12 @@ " - Each enum member represents an **attack technique** (the *how* of an attack)\n", " - Each 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 override `_prepare_strategies()` for custom composition logic (see `FoundryComposite`)\n", "\n", "2. **Scenario Class**: Extend `Scenario` and pass these to `super().__init__()`:\n", " - `strategy_class`: Your strategy enum class\n", " - `default_strategy`: The default strategy (typically `YourStrategy.ALL` or `YourStrategy.DEFAULT`)\n", - " - The base class provides a default `_get_atomic_attacks_async()` that uses the factory/registry\n", - " pattern. Override it only if your scenario needs custom attack construction logic.\n", + " - Implement `_build_atomic_attacks_async(context)` — the single abstract extension point.\n", + " Matrix-shaped scenarios delegate to `build_matrix_atomic_attacks(context=...)` in one line.\n", "\n", "3. **Default Dataset**: Pass `default_dataset_config=` to `super().__init__()` to specify the datasets your scenario uses out of the box.\n", " - Returns a `DatasetConfiguration` with one or more named datasets (e.g., `DatasetConfiguration(dataset_names=[\"my_dataset\"])`)\n", @@ -89,9 +88,10 @@ "\n", "### Example Structure\n", "\n", - "The simplest approach uses the **factory/registry pattern**: define your strategy,\n", - "dataset config, and constructor — the base class handles building atomic attacks\n", - "automatically from registered attack techniques." + "The construction path: define your strategy, dataset config, and constructor, then\n", + "implement `_build_atomic_attacks_async(context)`. Matrix-shaped scenarios delegate to the\n", + "`build_matrix_atomic_attacks` helper, which builds atomic attacks automatically from the\n", + "registered attack techniques." ] }, { @@ -124,6 +124,7 @@ " Scenario,\n", " ScenarioStrategy,\n", ")\n", + "from pyrit.scenario.core.matrix_atomic_attack_builder import build_matrix_atomic_attacks\n", "from pyrit.score.true_false.true_false_scorer import TrueFalseScorer\n", "from pyrit.setup import initialize_pyrit_async\n", "from pyrit.setup.initializers.components import ScenarioTechniqueInitializer\n", @@ -166,14 +167,15 @@ " scenario_result_id=scenario_result_id,\n", " )\n", "\n", - " # Optional: override _build_display_group to customize result grouping.\n", - " # Default groups by technique name; override to group by dataset instead:\n", - " def _build_display_group(self, *, technique_name: str, seed_group_name: str) -> str:\n", - " return seed_group_name\n", - "\n", - " # No _get_atomic_attacks_async override needed!\n", - " # The base class builds attacks from the (technique x dataset) cross-product\n", - " # using the factory/registry pattern automatically." + " # Implement the single abstract extension point. Matrix-shaped scenarios delegate\n", + " # to build_matrix_atomic_attacks; pass display_group_fn to customize result grouping\n", + " # (default groups by technique; here we group by dataset instead).\n", + " async def _build_atomic_attacks_async(self, *, context):\n", + " return build_matrix_atomic_attacks(\n", + " context=context,\n", + " objective_scorer=self._objective_scorer,\n", + " display_group_fn=lambda combo: combo.dataset_name,\n", + " )" ] }, { @@ -337,7 +339,7 @@ " ``TargetRegistry`` — typically by ``TargetInitializer`` from\n", " ``ADVERSARIAL_CHAT_*`` env vars, or programmatically via\n", " ``TargetRegistry.get_registry_singleton().instances.register``. At run\n", - " time, ``_get_atomic_attacks_async`` performs the ``(technique ×\n", + " time, ``_build_atomic_attacks_async`` performs the ``(technique ×\n", " adversarial_target × dataset)`` cross-product: for each selected\n", " adversarial-capable ``core`` factory in the ``AttackTechniqueRegistry``\n", " and each requested target, it calls\n", diff --git a/doc/code/scenarios/0_scenarios.py b/doc/code/scenarios/0_scenarios.py index 1b11a8dccb..2214a96ea4 100644 --- a/doc/code/scenarios/0_scenarios.py +++ b/doc/code/scenarios/0_scenarios.py @@ -59,13 +59,12 @@ # - Each enum member represents an **attack technique** (the *how* of an attack) # - Each 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 override `_prepare_strategies()` for custom composition logic (see `FoundryComposite`) # # 2. **Scenario Class**: Extend `Scenario` and pass these to `super().__init__()`: # - `strategy_class`: Your strategy enum class # - `default_strategy`: The default strategy (typically `YourStrategy.ALL` or `YourStrategy.DEFAULT`) -# - The base class provides a default `_get_atomic_attacks_async()` that uses the factory/registry -# pattern. Override it only if your scenario needs custom attack construction logic. +# - Implement `_build_atomic_attacks_async(context)` — the single abstract extension point. +# Matrix-shaped scenarios delegate to `build_matrix_atomic_attacks(context=...)` in one line. # # 3. **Default Dataset**: Pass `default_dataset_config=` to `super().__init__()` to specify the datasets your scenario uses out of the box. # - Returns a `DatasetConfiguration` with one or more named datasets (e.g., `DatasetConfiguration(dataset_names=["my_dataset"])`) @@ -91,9 +90,10 @@ # # ### Example Structure # -# The simplest approach uses the **factory/registry pattern**: define your strategy, -# dataset config, and constructor — the base class handles building atomic attacks -# automatically from registered attack techniques. +# The construction path: define your strategy, dataset config, and constructor, then +# implement `_build_atomic_attacks_async(context)`. Matrix-shaped scenarios delegate to the +# `build_matrix_atomic_attacks` helper, which builds atomic attacks automatically from the +# registered attack techniques. # %% from pyrit.common import apply_defaults @@ -102,6 +102,7 @@ Scenario, ScenarioStrategy, ) +from pyrit.scenario.core.matrix_atomic_attack_builder import build_matrix_atomic_attacks from pyrit.score.true_false.true_false_scorer import TrueFalseScorer from pyrit.setup import initialize_pyrit_async from pyrit.setup.initializers.components import ScenarioTechniqueInitializer @@ -144,14 +145,15 @@ def __init__( scenario_result_id=scenario_result_id, ) - # Optional: override _build_display_group to customize result grouping. - # Default groups by technique name; override to group by dataset instead: - def _build_display_group(self, *, technique_name: str, seed_group_name: str) -> str: - return seed_group_name - - # No _get_atomic_attacks_async override needed! - # The base class builds attacks from the (technique x dataset) cross-product - # using the factory/registry pattern automatically. + # Implement the single abstract extension point. Matrix-shaped scenarios delegate + # to build_matrix_atomic_attacks; pass display_group_fn to customize result grouping + # (default groups by technique; here we group by dataset instead). + async def _build_atomic_attacks_async(self, *, context): + return build_matrix_atomic_attacks( + context=context, + objective_scorer=self._objective_scorer, + display_group_fn=lambda combo: combo.dataset_name, + ) # %% [markdown] diff --git a/doc/scanner/1_pyrit_scan.ipynb b/doc/scanner/1_pyrit_scan.ipynb index 3fb5d1d1d9..0ec97066fc 100644 --- a/doc/scanner/1_pyrit_scan.ipynb +++ b/doc/scanner/1_pyrit_scan.ipynb @@ -242,9 +242,10 @@ " )\n", " # ... your scenario-specific initialization code\n", "\n", - " async def _get_atomic_attacks_async(self):\n", - " # Override only if your scenario needs custom attack construction.\n", - " # The base class provides a default that uses the factory/registry pattern.\n", + " async def _build_atomic_attacks_async(self, *, context):\n", + " # The single abstract extension point every scenario implements.\n", + " # Read runtime inputs from `context`; return the list of AtomicAttack to run.\n", + " # Matrix-shaped scenarios can delegate to build_matrix_atomic_attacks(context=...).\n", " # Example: create attacks for each strategy composite\n", " return []\n", "\n", diff --git a/doc/scanner/1_pyrit_scan.py b/doc/scanner/1_pyrit_scan.py index e42c2ddad6..fb22f97581 100644 --- a/doc/scanner/1_pyrit_scan.py +++ b/doc/scanner/1_pyrit_scan.py @@ -180,9 +180,10 @@ def __init__(self, *, scenario_result_id=None, **kwargs): ) # ... your scenario-specific initialization code - async def _get_atomic_attacks_async(self): - # Override only if your scenario needs custom attack construction. - # The base class provides a default that uses the factory/registry pattern. + async def _build_atomic_attacks_async(self, *, context): + # The single abstract extension point every scenario implements. + # Read runtime inputs from `context`; return the list of AtomicAttack to run. + # Matrix-shaped scenarios can delegate to build_matrix_atomic_attacks(context=...). # Example: create attacks for each strategy composite return [] diff --git a/pyrit/registry/components/scenario_registry.py b/pyrit/registry/components/scenario_registry.py index 2f28b87fe0..45dede7535 100644 --- a/pyrit/registry/components/scenario_registry.py +++ b/pyrit/registry/components/scenario_registry.py @@ -116,7 +116,7 @@ def _build_metadata(self, name: str, cls: type[Scenario]) -> ScenarioMetadata: Instantiates the scenario with no arguments and reads the strategy/dataset configuration off the instance. Every registered scenario MUST be no-arg instantiable (defer required-input validation to ``initialize_async`` or - ``_get_atomic_attacks_async``); otherwise this raises ``TypeError``. + ``_build_atomic_attacks_async``); otherwise this raises ``TypeError``. Args: name: The registry name of the scenario. @@ -140,7 +140,7 @@ def _build_metadata(self, name: str, cls: type[Scenario]) -> ScenarioMetadata: f"{name!r}) must be instantiable with no arguments so the registry can introspect " f"its strategies and default dataset config. Make all constructor parameters " f"optional (defaulting to None) and defer required-input validation to " - f"initialize_async() or _get_atomic_attacks_async(). Original error: {exc}" + f"initialize_async() or _build_atomic_attacks_async(). Original error: {exc}" ) from exc strategy_class = instance._strategy_class diff --git a/pyrit/scenario/core/attack_technique.py b/pyrit/scenario/core/attack_technique.py index 58525242dc..ff6e53ec7d 100644 --- a/pyrit/scenario/core/attack_technique.py +++ b/pyrit/scenario/core/attack_technique.py @@ -11,12 +11,7 @@ from typing import TYPE_CHECKING, Any -from pyrit.models import ( - AttackTechniqueIdentifier, - ComponentIdentifier, - Identifiable, - SeedIdentifier, -) +from pyrit.models import AttackTechniqueIdentifier, ComponentIdentifier, Identifiable, SeedIdentifier if TYPE_CHECKING: from pyrit.executor.attack import AttackStrategy diff --git a/pyrit/scenario/core/attack_technique_factory.py b/pyrit/scenario/core/attack_technique_factory.py index 5b64eb8a20..1912a4577f 100644 --- a/pyrit/scenario/core/attack_technique_factory.py +++ b/pyrit/scenario/core/attack_technique_factory.py @@ -29,11 +29,7 @@ from pyrit.common.deprecation import print_deprecation_message from pyrit.common.path import EXECUTOR_SEED_PROMPT_PATH from pyrit.executor.attack import PromptSendingAttack -from pyrit.executor.attack.core.attack_config import ( - AttackAdversarialConfig, - AttackConverterConfig, - AttackScoringConfig, -) +from pyrit.executor.attack.core.attack_config import AttackAdversarialConfig, AttackConverterConfig, AttackScoringConfig from pyrit.models import ( ComponentIdentifier, Identifiable, diff --git a/pyrit/scenario/core/dataset_configuration.py b/pyrit/scenario/core/dataset_configuration.py index 020a5a6964..4ea8db45b0 100644 --- a/pyrit/scenario/core/dataset_configuration.py +++ b/pyrit/scenario/core/dataset_configuration.py @@ -35,12 +35,7 @@ from pyrit.common.deprecation import print_deprecation_message from pyrit.memory import CentralMemory -from pyrit.models import ( - Seed, - SeedAttackGroup, - SeedGroup, - group_seeds_into_attack_groups, -) +from pyrit.models import Seed, SeedAttackGroup, SeedGroup, group_seeds_into_attack_groups if TYPE_CHECKING: from collections.abc import Callable, Sequence diff --git a/pyrit/scenario/core/matrix_atomic_attack_builder.py b/pyrit/scenario/core/matrix_atomic_attack_builder.py index 83c147f52a..3026c30a2c 100644 --- a/pyrit/scenario/core/matrix_atomic_attack_builder.py +++ b/pyrit/scenario/core/matrix_atomic_attack_builder.py @@ -123,7 +123,11 @@ def build_baseline_atomic_attack( ) -def resolve_technique_factories(*, context: ScenarioContext) -> dict[str, AttackTechniqueFactory]: +def resolve_technique_factories( + *, + context: ScenarioContext, + extra_factories: dict[str, AttackTechniqueFactory] | None = None, +) -> dict[str, AttackTechniqueFactory]: """ Resolve a run's selected strategies to their registered ``AttackTechniqueFactory`` instances. @@ -133,6 +137,10 @@ def resolve_technique_factories(*, context: ScenarioContext) -> dict[str, Attack Args: context (ScenarioContext): The resolved runtime inputs for this run. + extra_factories (dict[str, AttackTechniqueFactory] | None): Scenario-local factories + merged on top of the registry before filtering, so a scenario can offer techniques + without registering them globally. Entries override registry factories of the same + name. Returns: dict[str, AttackTechniqueFactory]: Mapping of technique name to factory, ordered by @@ -140,7 +148,9 @@ def resolve_technique_factories(*, context: ScenarioContext) -> dict[str, Attack """ from pyrit.registry.components.attack_technique_registry import AttackTechniqueRegistry - all_factories = AttackTechniqueRegistry.get_registry_singleton().get_factories_or_raise() + all_factories = dict(AttackTechniqueRegistry.get_registry_singleton().get_factories_or_raise()) + if extra_factories: + all_factories.update(extra_factories) return { strategy.value: all_factories[strategy.value] for strategy in context.scenario_strategies @@ -154,6 +164,7 @@ def build_matrix_atomic_attacks( objective_scorer: Scorer, display_group_fn: Callable[[MatrixCombo], str] | None = None, strategy_converters: dict[str, list[PromptConverter]] | None = None, + extra_factories: dict[str, AttackTechniqueFactory] | None = None, ) -> list[AtomicAttack]: """ Build a matrix-shaped scenario's atomic attacks from its resolved context in one call. @@ -162,7 +173,7 @@ def build_matrix_atomic_attacks( technique × dataset cross-product: it resolves the selected strategies to factories (``resolve_technique_factories``) and hands them to ``MatrixAtomicAttackBuilder`` with the context's target, labels, and per-dataset seed groups. The baseline is emitted - centrally by ``Scenario._get_atomic_attacks_async``, so this never prepends one. + centrally by ``Scenario.initialize_async``, so this never prepends one. Scenarios needing extra axes (adversarial targets, caching, converter stacks) call ``MatrixAtomicAttackBuilder`` directly instead. @@ -176,6 +187,9 @@ def build_matrix_atomic_attacks( technique name to converters appended after that technique's converters. Pass a scenario's ``self._strategy_converters`` so per-technique converter overrides are preserved. + extra_factories (dict[str, AttackTechniqueFactory] | None): Scenario-local factories + merged on top of the registry (see ``resolve_technique_factories``), so a scenario + can offer techniques without registering them globally. Returns: list[AtomicAttack]: The generated atomic attacks (no baseline). @@ -186,7 +200,7 @@ def build_matrix_atomic_attacks( memory_labels=context.memory_labels, ) return builder.build( - technique_factories=resolve_technique_factories(context=context), + technique_factories=resolve_technique_factories(context=context, extra_factories=extra_factories), dataset_groups=context.seed_groups_by_dataset, display_group_fn=display_group_fn, strategy_converters=strategy_converters, diff --git a/pyrit/scenario/core/scenario.py b/pyrit/scenario/core/scenario.py index de59d2c004..8556e25e2e 100644 --- a/pyrit/scenario/core/scenario.py +++ b/pyrit/scenario/core/scenario.py @@ -11,7 +11,7 @@ import asyncio import logging import uuid -from abc import ABC +from abc import ABC, abstractmethod from collections.abc import Sequence from enum import Enum from pathlib import Path @@ -27,7 +27,6 @@ from tqdm.auto import tqdm from pyrit.common import REQUIRED_VALUE, apply_defaults -from pyrit.common.deprecation import print_deprecation_message from pyrit.common.utils import to_sha256 from pyrit.executor.attack import AttackExecutor from pyrit.memory import CentralMemory @@ -45,15 +44,10 @@ from pyrit.prompt_target import PromptTarget from pyrit.prompt_target.common.target_requirements import TargetRequirements from pyrit.registry import ScorerRegistry -from pyrit.registry.resolution import ( - resolve_declared_params, -) +from pyrit.registry.resolution import resolve_declared_params from pyrit.scenario.core.atomic_attack import AtomicAttack from pyrit.scenario.core.dataset_configuration import DatasetAttackConfiguration -from pyrit.scenario.core.matrix_atomic_attack_builder import ( - MatrixAtomicAttackBuilder, - build_baseline_atomic_attack, -) +from pyrit.scenario.core.matrix_atomic_attack_builder import build_baseline_atomic_attack from pyrit.scenario.core.scenario_context import ScenarioContext from pyrit.scenario.core.scenario_strategy import ScenarioStrategy from pyrit.scenario.core.scenario_target_defaults import get_default_scorer_target @@ -70,7 +64,6 @@ if TYPE_CHECKING: from pyrit.models import ComponentIdentifier from pyrit.prompt_converter import PromptConverter - from pyrit.scenario.core.attack_technique_factory import AttackTechniqueFactory logger = logging.getLogger(__name__) @@ -103,7 +96,7 @@ class BaselineAttackPolicy(Enum): Forbidden = "forbidden" -class Scenario(ABC): # noqa: B024 - retained for subclass type-checking even without abstract methods +class Scenario(ABC): """ Groups and executes multiple AtomicAttack instances sequentially. @@ -163,7 +156,6 @@ def __init__( default_dataset_config: DatasetAttackConfiguration, objective_scorer: Scorer, scenario_result_id: uuid.UUID | str | None = None, - include_default_baseline: bool | None = None, # Deprecated. Will be removed in 0.16.0. ) -> None: """ Initialize a scenario. @@ -182,14 +174,10 @@ def __init__( Can be either a UUID object or a string representation of a UUID. If provided and found in memory, the scenario will resume from prior progress. All other parameters must still match the stored scenario configuration. - include_default_baseline (bool | None): **Deprecated.** Will be removed in 0.16.0. - Pass ``include_baseline`` to ``initialize_async`` instead. When set, the value is - used as the effective ``include_baseline`` for the next ``initialize_async`` call - unless that call passes its own ``include_baseline``. Note: Attack runs are populated by calling initialize_async(), which invokes the - subclass's _get_atomic_attacks_async() method. + subclass's _build_atomic_attacks_async() method. The scenario description is automatically extracted from the class's docstring (__doc__) with whitespace normalized for display. @@ -231,7 +219,7 @@ def __init__( self._atomic_attacks: list[AtomicAttack] = [] self._scenario_result_id: str | None = str(scenario_result_id) if scenario_result_id else None - # Store prepared strategies for use in _get_atomic_attacks_async + # Store prepared strategies for use in _build_atomic_attacks_async self._scenario_strategies: list[ScenarioStrategy] = [] # Maps concrete technique name → extra request converters to append for that technique. @@ -241,25 +229,13 @@ def __init__( self._display_group_map: dict[str, str] = {} # Declared via supported_parameters(); resolved/populated by the registry - # helper (pyrit.registry.resolution). Subclasses read it in _get_atomic_attacks_async. + # helper (pyrit.registry.resolution). Subclasses read it in _build_atomic_attacks_async. self.params: dict[str, Any] = {} # Resolved effective baseline inclusion for the current run. Set in initialize_async - # before _get_atomic_attacks_async is awaited so overrides can read it. + # before _build_atomic_attacks_async is awaited so overrides can read it. self._include_baseline: bool = False - # Deprecated constructor-time baseline override. Will be removed in 0.16.0, along - # with the include_default_baseline kwarg above and the legacy fallback branch in - # initialize_async. Subclass shims set this attribute directly to avoid double-warning. - self._legacy_include_baseline: bool | None = None - if include_default_baseline is not None: - print_deprecation_message( - old_item="Scenario(include_default_baseline=...)", - new_item="Scenario.initialize_async(include_baseline=...)", - removed_in="0.16.0", - ) - self._legacy_include_baseline = include_default_baseline - @property def name(self) -> str: """The name of the scenario.""" @@ -287,64 +263,6 @@ def supported_parameters(cls) -> list[Parameter]: """ return [] - def _get_attack_technique_factories(self) -> dict[str, "AttackTechniqueFactory"]: - """ - Return the attack technique factories for this scenario. - - Each key is a technique name (matching a strategy enum value) and each - value is an ``AttackTechniqueFactory`` that can produce an - ``AttackTechnique`` for that technique. - - The base implementation returns every factory currently registered in - the ``AttackTechniqueRegistry`` singleton. The canonical scenario - techniques are populated by ``ScenarioTechniqueInitializer`` - (``pyrit.setup.initializers.components.scenario_techniques``); ensure - that initializer has run before scenarios use this method. - Subclasses may override to add, remove, or replace factories. - - Returns: - dict[str, AttackTechniqueFactory]: Mapping of technique name to factory. - - Raises: - RuntimeError: If the registry is empty (no initializer has run). - """ - from pyrit.registry.components.attack_technique_registry import AttackTechniqueRegistry - - registry = AttackTechniqueRegistry.get_registry_singleton() - return registry.get_factories_or_raise() - - def _build_display_group(self, *, technique_name: str, seed_group_name: str) -> str: - """ - Build the display-group label for an atomic attack. - - Each ``AtomicAttack`` has a unique ``atomic_attack_name`` (e.g. - ``"prompt_sending_airt_hate"``) used for resume tracking. However, - user-facing output (console printer, reports) often needs to - aggregate results along a *different* dimension — for example, - grouping by harm category rather than by technique. The display - group provides that second grouping axis without affecting resume - behaviour. - - The default groups by technique name. Subclasses override to - change the aggregation axis: - - - **By technique** (default): ``return technique_name`` - - **By harm category / dataset**: ``return seed_group_name`` - - **Cross-product**: ``return f"{technique_name}_{seed_group_name}"`` - - Note: ``seed_group_name`` is the dataset key from - ``DatasetAttackConfiguration.get_attack_groups_by_dataset_async()`` (e.g. - ``"airt_hate"``), not a ``SeedGroup`` object. - - Args: - technique_name: The name of the attack technique. - seed_group_name: The dataset key from the dataset configuration. - - Returns: - str: The display-group label. - """ - return technique_name - def _get_default_objective_scorer(self) -> TrueFalseScorer: # Deferred import to avoid circular dependency. from pyrit.setup.initializers.components.scorers import ScorerInitializerTags @@ -433,28 +351,6 @@ def set_params_from_args(self, *, args: dict[str, Any]) -> None: owner=f"Scenario '{type(self).__name__}'", ) - def _prepare_strategies( - self, - strategies: Sequence[ScenarioStrategy] | None, - ) -> 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 (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 or [] both mean use - default; otherwise a list of strategies to resolve. - - Returns: - list[ScenarioStrategy]: Ordered, deduplicated concrete strategies. - """ - return self._strategy_class.resolve(strategies, default=self._default_strategy) - @apply_defaults async def initialize_async( self, @@ -535,12 +431,6 @@ async def initialize_async( self._max_retries = max_retries self._memory_labels = memory_labels or {} - # Deprecated. Will be removed in 0.16.0. Honor the legacy constructor-time - # include_default_baseline (or subclass include_baseline) only when the caller did - # not supply a runtime value. - if include_baseline is None and self._legacy_include_baseline is not None: - include_baseline = self._legacy_include_baseline - # Resolve the effective include_baseline. Forbidden is checked first so a forbidden # scenario type never silently inherits a True default; explicit-True on a forbidden # type is a hard error rather than a silent ignore. For the Enabled / Disabled states, @@ -558,7 +448,7 @@ async def initialize_async( self._include_baseline = include_baseline # Prepare scenario strategies using the stored configuration - self._scenario_strategies = self._prepare_strategies(scenario_strategies) + self._scenario_strategies = self._strategy_class.resolve(scenario_strategies, default=self._default_strategy) self._strategy_converters = strategy_converters or {} # Resolve declared parameters through the single registry-owned path, @@ -567,25 +457,16 @@ async def initialize_async( # so the registry- and CLI-driven flows converge here without divergence. self.set_params_from_args(args=self.params) - self._atomic_attacks = await self._get_atomic_attacks_async() + # Build atomic attacks: resolve the seed groups once, snapshot the resolved inputs + # into a ScenarioContext, and hand it to the subclass extension point. Baseline is + # emitted centrally (from context.seed_groups) so overrides never re-resolve seeds + # or hand-roll baseline emission. + seed_groups_by_dataset = await self._resolve_seed_groups_by_dataset_async() + context = self._build_scenario_context(seed_groups_by_dataset=seed_groups_by_dataset) + self._atomic_attacks = await self._build_atomic_attacks_async(context=context) - # Deprecation rescue. Will be removed in 0.16.0. If the override didn't emit baseline, - # warn and inject. Migrated overrides emit baseline themselves and bypass this branch. - # Reuse seeds from the first existing attack rather than re-resolving from - # dataset_config; re-resolution under max_dataset_size would draw a fresh sample - # (the very ADO 9012 bug this PR fixes). When no atomic attacks exist yet the - # rescue falls back to the dataset_config one-time resolution. if include_baseline and (not self._atomic_attacks or self._atomic_attacks[0].atomic_attack_name != "baseline"): - print_deprecation_message( - old_item=f"Implicit baseline injection for {type(self).__name__}._get_atomic_attacks_async()", - new_item="explicit emission via self._build_baseline_atomic_attack(seed_groups=...) in the override", - removed_in="0.16.0", - ) - if self._atomic_attacks: - seed_groups = self._atomic_attacks[0].seed_groups - else: - seed_groups = await self._dataset_config.get_seed_attack_groups_async() - self._atomic_attacks.insert(0, self._build_baseline_atomic_attack(seed_groups=seed_groups)) + self._atomic_attacks.insert(0, self._build_baseline_atomic_attack(seed_groups=list(context.seed_groups))) # Build the canonical scenario identifier once params/strategies/datasets # are resolved, so both the resume check and the new-result branch share the @@ -945,47 +826,19 @@ def _build_scenario_context(self, *, seed_groups_by_dataset: dict[str, list[Seed seed_groups_by_dataset=seed_groups_by_dataset, ) - async def _get_atomic_attacks_async(self) -> list[AtomicAttack]: - """ - Build this scenario's atomic attacks (internal entry point called by ``initialize_async``). - - Resolves the seed groups once, builds a ``ScenarioContext`` from the values resolved - in ``initialize_async``, and forwards to ``_build_atomic_attacks_async`` — the extension - point scenarios override to customize attack construction. The baseline is emitted - centrally here (when ``context.include_baseline`` is set) from ``context.seed_groups``, - so overrides never re-resolve seeds or hand-roll baseline emission. This stays a stable, - no-argument entry point for ``initialize_async`` and other internal callers. - - Returns: - list[AtomicAttack]: The generated atomic attacks. - - Raises: - ValueError: If the scenario has not been initialized. - """ - seed_groups_by_dataset = await self._resolve_seed_groups_by_dataset_async() - context = self._build_scenario_context(seed_groups_by_dataset=seed_groups_by_dataset) - atomic_attacks = await self._build_atomic_attacks_async(context=context) - - # Central baseline emission. Guarded so a scenario that still emits its own baseline - # (or an aggregate that legitimately has none) isn't given a duplicate. - if context.include_baseline and (not atomic_attacks or atomic_attacks[0].atomic_attack_name != "baseline"): - atomic_attacks.insert(0, self._build_baseline_atomic_attack(seed_groups=list(context.seed_groups))) - - return atomic_attacks - + @abstractmethod async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list[AtomicAttack]: """ - Build atomic attacks from the cross-product of selected techniques and datasets. + Build this scenario's atomic attacks from the resolved runtime inputs. This is the single extension point scenarios override to map techniques, datasets, - scorers, and any extra axes into ``AtomicAttack`` instances. The default - implementation delegates to ``MatrixAtomicAttackBuilder`` using the - ``_get_attack_technique_factories()`` and ``_build_display_group()`` hooks, producing - one ``AtomicAttack`` per (technique × dataset) pair. - - Scenarios with custom construction (composite attacks, per-objective technique - selection, converter stacks) override this method and build their attacks from - ``context.seed_groups`` (or ``context.seed_groups_by_dataset``). The base owns baseline + scorers, and any extra axes into ``AtomicAttack`` instances. It is called once by + ``initialize_async`` after the objective target, scorer, strategies, dataset config, + labels, and baseline flag have been resolved and snapshot into ``context``. + + Scenario authors build their attacks from ``context.seed_groups`` (or + ``context.seed_groups_by_dataset``) so sampling under ``max_dataset_size`` stays + consistent across every atomic attack and the baseline. The base owns baseline emission, so overrides never prepend one themselves. Args: @@ -994,32 +847,7 @@ async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list Returns: list[AtomicAttack]: The generated atomic attacks. """ - selected_techniques = {s.value for s in context.scenario_strategies} - all_factories = self._get_attack_technique_factories() - - technique_factories: dict[str, AttackTechniqueFactory] = {} - for technique_name in selected_techniques: - factory = all_factories.get(technique_name) - if factory is None: - logger.warning(f"No factory for technique '{technique_name}', skipping.") - continue - technique_factories[technique_name] = factory - - builder = MatrixAtomicAttackBuilder( - objective_target=context.objective_target, - objective_scorer=self._objective_scorer, - memory_labels=context.memory_labels, - ) - return builder.build( - technique_factories=technique_factories, - dataset_groups=context.seed_groups_by_dataset, - display_group_fn=lambda combo: self._build_display_group( - technique_name=combo.technique_name, - seed_group_name=combo.dataset_name, - ), - strategy_converters=self._strategy_converters, - include_baseline=False, - ) + ... async def run_async(self) -> ScenarioResult: """ diff --git a/pyrit/scenario/scenarios/adaptive/__init__.py b/pyrit/scenario/scenarios/adaptive/__init__.py index 98bdf444dc..4be199024c 100644 --- a/pyrit/scenario/scenarios/adaptive/__init__.py +++ b/pyrit/scenario/scenarios/adaptive/__init__.py @@ -9,11 +9,7 @@ AdaptiveTechniqueDispatcher, TechniqueBundle, ) -from pyrit.scenario.scenarios.adaptive.selectors import ( - EpsilonGreedyTechniqueSelector, - SelectorScope, - TechniqueSelector, -) +from pyrit.scenario.scenarios.adaptive.selectors import EpsilonGreedyTechniqueSelector, SelectorScope, TechniqueSelector from pyrit.scenario.scenarios.adaptive.text_adaptive import TextAdaptive __all__ = [ diff --git a/pyrit/scenario/scenarios/adaptive/adaptive_scenario.py b/pyrit/scenario/scenarios/adaptive/adaptive_scenario.py index 39ebb4ee91..a4abfe916d 100644 --- a/pyrit/scenario/scenarios/adaptive/adaptive_scenario.py +++ b/pyrit/scenario/scenarios/adaptive/adaptive_scenario.py @@ -27,14 +27,8 @@ from pyrit.scenario.core.attack_technique import AttackTechnique from pyrit.scenario.core.scenario import Scenario from pyrit.scenario.core.scenario_target_defaults import get_default_adversarial_target -from pyrit.scenario.scenarios.adaptive.dispatcher import ( - AdaptiveTechniqueDispatcher, - TechniqueBundle, -) -from pyrit.scenario.scenarios.adaptive.selectors import ( - EpsilonGreedyTechniqueSelector, - TechniqueSelector, -) +from pyrit.scenario.scenarios.adaptive.dispatcher import AdaptiveTechniqueDispatcher, TechniqueBundle +from pyrit.scenario.scenarios.adaptive.selectors import EpsilonGreedyTechniqueSelector, TechniqueSelector if TYPE_CHECKING: from pyrit.models import SeedAttackGroup @@ -146,13 +140,12 @@ def _get_attack_technique_factories(self) -> dict[str, AttackTechniqueFactory]: # Local import: ``scenario_techniques`` imports ``pyrit.scenario.core``, # which transitively re-imports this module, so a top-level import # would form a cycle during ``pyrit.scenario`` package initialization. - from pyrit.setup.initializers.components.scenario_techniques import ( - build_scenario_technique_factories, - ) + from pyrit.registry.components.attack_technique_registry import AttackTechniqueRegistry + from pyrit.setup.initializers.components.scenario_techniques import build_scenario_technique_factories catalog = {factory.name: factory for factory in build_scenario_technique_factories()} try: - registry_overrides = super()._get_attack_technique_factories() + registry_overrides = AttackTechniqueRegistry.get_registry_singleton().get_factories_or_raise() except RuntimeError: # Registry not initialized yet (e.g. bare CLI parse before # ScenarioTechniqueInitializer has run). Catalog alone is the diff --git a/pyrit/scenario/scenarios/adaptive/selectors/__init__.py b/pyrit/scenario/scenarios/adaptive/selectors/__init__.py index 709146fb43..50aa4e828a 100644 --- a/pyrit/scenario/scenarios/adaptive/selectors/__init__.py +++ b/pyrit/scenario/scenarios/adaptive/selectors/__init__.py @@ -3,13 +3,8 @@ """Selector protocol and selector implementations.""" -from pyrit.scenario.scenarios.adaptive.selectors.epsilon_greedy import ( - EpsilonGreedyTechniqueSelector, -) -from pyrit.scenario.scenarios.adaptive.selectors.technique_selector import ( - SelectorScope, - TechniqueSelector, -) +from pyrit.scenario.scenarios.adaptive.selectors.epsilon_greedy import EpsilonGreedyTechniqueSelector +from pyrit.scenario.scenarios.adaptive.selectors.technique_selector import SelectorScope, TechniqueSelector __all__ = [ "EpsilonGreedyTechniqueSelector", diff --git a/pyrit/scenario/scenarios/adaptive/text_adaptive.py b/pyrit/scenario/scenarios/adaptive/text_adaptive.py index 492694607f..a866fba125 100644 --- a/pyrit/scenario/scenarios/adaptive/text_adaptive.py +++ b/pyrit/scenario/scenarios/adaptive/text_adaptive.py @@ -18,14 +18,9 @@ from pyrit.common import apply_defaults from pyrit.models.parameter import Parameter -from pyrit.registry.components.attack_technique_registry import ( - AttackTechniqueRegistry, -) +from pyrit.registry.components.attack_technique_registry import AttackTechniqueRegistry from pyrit.registry.tag_query import TagQuery -from pyrit.scenario.core.dataset_configuration import ( - CompoundDatasetAttackConfiguration, - DatasetAttackConfiguration, -) +from pyrit.scenario.core.dataset_configuration import CompoundDatasetAttackConfiguration, DatasetAttackConfiguration from pyrit.scenario.scenarios.adaptive.adaptive_scenario import AdaptiveScenario if TYPE_CHECKING: @@ -57,9 +52,7 @@ def _build_text_adaptive_strategy() -> type[ScenarioStrategy]: # Local import: ``scenario_techniques`` imports ``pyrit.scenario.core``, # which transitively re-imports this module, so a top-level import would # form a cycle during ``pyrit.scenario`` package initialization. - from pyrit.setup.initializers.components.scenario_techniques import ( - build_scenario_technique_factories, - ) + from pyrit.setup.initializers.components.scenario_techniques import build_scenario_technique_factories all_factories = list(build_scenario_technique_factories()) catalog_names = {factory.name for factory in all_factories} diff --git a/pyrit/scenario/scenarios/airt/cyber.py b/pyrit/scenario/scenarios/airt/cyber.py index 564ae4ef5d..b8923d31bc 100644 --- a/pyrit/scenario/scenarios/airt/cyber.py +++ b/pyrit/scenario/scenarios/airt/cyber.py @@ -8,7 +8,6 @@ from typing import TYPE_CHECKING from pyrit.common import apply_defaults -from pyrit.common.deprecation import print_deprecation_message # Deprecated. Will be removed in 0.16.0. from pyrit.common.path import SCORER_SEED_PROMPT_PATH from pyrit.scenario.core.dataset_configuration import DatasetAttackConfiguration from pyrit.scenario.core.matrix_atomic_attack_builder import build_matrix_atomic_attacks @@ -83,7 +82,6 @@ def __init__( *, objective_scorer: TrueFalseScorer | None = None, scenario_result_id: str | None = None, - include_baseline: bool | None = None, # Deprecated. Will be removed in 0.16.0. ) -> None: """ Initialize the cyber harms scenario. @@ -92,8 +90,6 @@ def __init__( objective_scorer (TrueFalseScorer | None): Objective scorer for malware detection. If not provided, defaults to a composite scorer using malware detection + refusal backstop. scenario_result_id (str | None): Optional ID of an existing scenario result to resume. - include_baseline (bool | None): **Deprecated.** Will be removed in 0.16.0. Pass - ``include_baseline`` to ``initialize_async`` instead. """ self._objective_scorer: TrueFalseScorer = ( objective_scorer if objective_scorer else self._get_default_objective_scorer() @@ -110,22 +106,12 @@ def __init__( scenario_result_id=scenario_result_id, ) - # Deprecated constructor-time baseline override. Will be removed in 0.16.0, along with - # the include_baseline kwarg above. - if include_baseline is not None: - print_deprecation_message( - old_item="Cyber(include_baseline=...)", - new_item="Cyber.initialize_async(include_baseline=...)", - removed_in="0.16.0", - ) - self._legacy_include_baseline = include_baseline - async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list[AtomicAttack]: """ Build the technique × dataset atomic attacks for Cyber, grouped by technique. - The baseline is emitted centrally by the base ``_get_atomic_attacks_async`` bridge, so - this override never prepends one. + The baseline is emitted centrally by the base ``initialize_async``, so this override + never prepends one. Args: context (ScenarioContext): The resolved runtime inputs for this run. diff --git a/pyrit/scenario/scenarios/airt/jailbreak.py b/pyrit/scenario/scenarios/airt/jailbreak.py index 9299a5406a..7ed3bf2efb 100644 --- a/pyrit/scenario/scenarios/airt/jailbreak.py +++ b/pyrit/scenario/scenarios/airt/jailbreak.py @@ -5,13 +5,8 @@ from typing import Any from pyrit.common import apply_defaults -from pyrit.common.deprecation import print_deprecation_message # Deprecated. Will be removed in 0.16.0. from pyrit.datasets import TextJailBreak -from pyrit.executor.attack.core.attack_config import ( - AttackAdversarialConfig, - AttackConverterConfig, - AttackScoringConfig, -) +from pyrit.executor.attack.core.attack_config import AttackAdversarialConfig, AttackConverterConfig, AttackScoringConfig from pyrit.executor.attack.single_turn.many_shot_jailbreak import ManyShotJailbreakAttack from pyrit.executor.attack.single_turn.prompt_sending import PromptSendingAttack from pyrit.executor.attack.single_turn.role_play import RolePlayAttack, RolePlayPaths @@ -22,16 +17,12 @@ from pyrit.prompt_target.common.prompt_target import PromptTarget from pyrit.scenario.core.atomic_attack import AtomicAttack from pyrit.scenario.core.attack_technique import AttackTechnique -from pyrit.scenario.core.dataset_configuration import ( - DatasetAttackConfiguration, -) +from pyrit.scenario.core.dataset_configuration import DatasetAttackConfiguration from pyrit.scenario.core.scenario import Scenario from pyrit.scenario.core.scenario_context import ScenarioContext from pyrit.scenario.core.scenario_strategy import ScenarioStrategy from pyrit.scenario.core.scenario_target_defaults import get_default_adversarial_target -from pyrit.score import ( - TrueFalseScorer, -) +from pyrit.score import TrueFalseScorer class JailbreakStrategy(ScenarioStrategy): @@ -98,7 +89,6 @@ def __init__( num_templates: int | None = None, num_attempts: int = 1, jailbreak_names: list[str] | None = None, - include_baseline: bool | None = None, # Deprecated. Will be removed in 0.16.0. ) -> None: """ Initialize the jailbreak scenario. @@ -111,8 +101,6 @@ def __init__( num_attempts (int | None): Number of times to try each jailbreak. jailbreak_names (list[str] | None): List of jailbreak names from the template list under datasets. to use. - include_baseline (bool | None): **Deprecated.** Will be removed in 0.16.0. Pass - ``include_baseline`` to ``initialize_async`` instead. Raises: ValueError: If both jailbreak_names and num_templates are provided, as random selection @@ -162,16 +150,6 @@ def __init__( scenario_result_id=scenario_result_id, ) - # Deprecated constructor-time baseline override. Will be removed in 0.16.0, along with - # the include_baseline kwarg above. - if include_baseline is not None: - print_deprecation_message( - old_item="Jailbreak(include_baseline=...)", - new_item="Jailbreak.initialize_async(include_baseline=...)", - removed_in="0.16.0", - ) - self._legacy_include_baseline = include_baseline - def _get_or_create_adversarial_target(self) -> PromptTarget: """ Return the shared adversarial target, creating it on first access. diff --git a/pyrit/scenario/scenarios/airt/leakage.py b/pyrit/scenario/scenarios/airt/leakage.py index 8e75041232..4346624a65 100644 --- a/pyrit/scenario/scenarios/airt/leakage.py +++ b/pyrit/scenario/scenarios/airt/leakage.py @@ -9,24 +9,22 @@ from pyrit.common import apply_defaults from pyrit.common.path import DATASETS_PATH, SCORER_SEED_PROMPT_PATH -from pyrit.executor.attack import ( - AttackConverterConfig, - PromptSendingAttack, -) +from pyrit.executor.attack import AttackConverterConfig, PromptSendingAttack from pyrit.prompt_converter import AddImageTextConverter, FirstLetterConverter from pyrit.prompt_normalizer import PromptConverterConfiguration -from pyrit.registry.components.attack_technique_registry import ( - AttackTechniqueRegistry, -) +from pyrit.registry.components.attack_technique_registry import AttackTechniqueRegistry from pyrit.registry.tag_query import TagQuery from pyrit.scenario.core.attack_technique_factory import AttackTechniqueFactory from pyrit.scenario.core.dataset_configuration import DatasetAttackConfiguration +from pyrit.scenario.core.matrix_atomic_attack_builder import build_matrix_atomic_attacks from pyrit.scenario.core.scenario import Scenario from pyrit.scenario.core.scenario_strategy import ScenarioStrategy if TYPE_CHECKING: from pathlib import Path + from pyrit.scenario.core.atomic_attack import AtomicAttack + from pyrit.scenario.core.scenario_context import ScenarioContext from pyrit.scenario.core.scenario_strategy import ScenarioStrategy from pyrit.score import TrueFalseScorer @@ -144,20 +142,24 @@ def __init__( scenario_result_id=scenario_result_id, ) - def _get_attack_technique_factories(self) -> dict[str, AttackTechniqueFactory]: + async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list[AtomicAttack]: """ - Return core + leakage-specific attack technique factories. + Build the Leakage atomic attacks from the selected core + leakage techniques. - Gets core factories from the base class, then merges in the - leakage-specific factories (kept local to this scenario so they don't - pollute the global registry). + Passes the leakage-specific factories (``first_letter``, ``image``) as + ``extra_factories`` — kept local to this scenario so they don't pollute the global + registry — and delegates the technique × dataset cross-product to + ``build_matrix_atomic_attacks``. The base owns baseline emission. + + Args: + context (ScenarioContext): The resolved runtime inputs for this run. Returns: - dict[str, AttackTechniqueFactory]: Mapping of technique names to their factories. + list[AtomicAttack]: The generated atomic attacks. """ - factories = super()._get_attack_technique_factories() - - for factory in LEAKAGE_FACTORIES: - factories[factory.name] = factory - - return factories + return build_matrix_atomic_attacks( + context=context, + objective_scorer=self._objective_scorer, + strategy_converters=self._strategy_converters, + extra_factories={factory.name: factory for factory in LEAKAGE_FACTORIES}, + ) diff --git a/pyrit/scenario/scenarios/airt/psychosocial.py b/pyrit/scenario/scenarios/airt/psychosocial.py index 678e99bb93..18b89841b2 100644 --- a/pyrit/scenario/scenarios/airt/psychosocial.py +++ b/pyrit/scenario/scenarios/airt/psychosocial.py @@ -9,7 +9,6 @@ import yaml from pyrit.common import apply_defaults -from pyrit.common.deprecation import print_deprecation_message # Deprecated. Will be removed in 0.16.0. from pyrit.common.path import DATASETS_PATH from pyrit.executor.attack import ( AttackAdversarialConfig, @@ -23,22 +22,15 @@ ) from pyrit.models import SeedAttackGroup, SeedObjective, SeedPrompt from pyrit.prompt_converter import ToneConverter -from pyrit.prompt_normalizer.prompt_converter_configuration import ( - PromptConverterConfiguration, -) +from pyrit.prompt_normalizer.prompt_converter_configuration import PromptConverterConfiguration from pyrit.prompt_target import CapabilityName, PromptTarget from pyrit.prompt_target.common.target_requirements import CHAT_TARGET_REQUIREMENTS, TargetRequirements from pyrit.scenario.core.atomic_attack import AtomicAttack from pyrit.scenario.core.attack_technique import AttackTechnique -from pyrit.scenario.core.dataset_configuration import ( - DatasetAttackConfiguration, - DatasetConstraintError, -) +from pyrit.scenario.core.dataset_configuration import DatasetAttackConfiguration, DatasetConstraintError from pyrit.scenario.core.scenario import Scenario from pyrit.scenario.core.scenario_context import ScenarioContext -from pyrit.scenario.core.scenario_strategy import ( - ScenarioStrategy, -) +from pyrit.scenario.core.scenario_strategy import ScenarioStrategy from pyrit.scenario.core.scenario_target_defaults import get_default_adversarial_target, get_default_scorer_target from pyrit.score import ( FloatScaleScorer, @@ -181,7 +173,6 @@ def __init__( scenario_result_id: str | None = None, subharm_configs: dict[str, SubharmConfig] | None = None, max_turns: int = 5, - include_baseline: bool | None = None, # Deprecated. Will be removed in 0.16.0. ) -> None: """ Initialize the Psychosocial Harms Scenario. @@ -213,8 +204,6 @@ def __init__( max_turns (int): Maximum number of conversation turns for multi-turn attacks (CrescendoAttack). Defaults to 5. Increase for more gradual escalation, decrease for faster testing. - include_baseline (bool | None): **Deprecated.** Will be removed in 0.16.0. Pass - ``include_baseline`` to ``initialize_async`` instead. """ if objectives is not None: logger.warning( @@ -240,16 +229,6 @@ def __init__( scenario_result_id=scenario_result_id, ) - # Deprecated constructor-time baseline override. Will be removed in 0.16.0, along with - # the include_baseline kwarg above. - if include_baseline is not None: - print_deprecation_message( - old_item="Psychosocial(include_baseline=...)", - new_item="Psychosocial.initialize_async(include_baseline=...)", - removed_in="0.16.0", - ) - self._legacy_include_baseline = include_baseline - # Store deprecated objectives for later resolution in _resolve_seed_groups_by_dataset_async self._deprecated_objectives = objectives diff --git a/pyrit/scenario/scenarios/airt/rapid_response.py b/pyrit/scenario/scenarios/airt/rapid_response.py index 67887a783c..652a1e9711 100644 --- a/pyrit/scenario/scenarios/airt/rapid_response.py +++ b/pyrit/scenario/scenarios/airt/rapid_response.py @@ -117,7 +117,7 @@ async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list Results group by harm category (the dataset name) rather than technique so per-category ASR rolls up naturally. The baseline is emitted centrally by the base - ``_get_atomic_attacks_async`` bridge, so this override never prepends one. + ``initialize_async``, so this override never prepends one. Args: context (ScenarioContext): The resolved runtime inputs for this run. diff --git a/pyrit/scenario/scenarios/airt/scam.py b/pyrit/scenario/scenarios/airt/scam.py index 6c8fa05b9d..61db79dd71 100644 --- a/pyrit/scenario/scenarios/airt/scam.py +++ b/pyrit/scenario/scenarios/airt/scam.py @@ -6,28 +6,14 @@ from typing import TYPE_CHECKING, Any from pyrit.common import apply_defaults -from pyrit.common.deprecation import print_deprecation_message # Deprecated. Will be removed in 0.16.0. -from pyrit.common.path import ( - EXECUTOR_RED_TEAM_PATH, - SCORER_SEED_PROMPT_PATH, -) -from pyrit.executor.attack import ( - ContextComplianceAttack, - RedTeamingAttack, - RolePlayAttack, - RolePlayPaths, -) -from pyrit.executor.attack.core.attack_config import ( - AttackAdversarialConfig, - AttackScoringConfig, -) +from pyrit.common.path import EXECUTOR_RED_TEAM_PATH, SCORER_SEED_PROMPT_PATH +from pyrit.executor.attack import ContextComplianceAttack, RedTeamingAttack, RolePlayAttack, RolePlayPaths +from pyrit.executor.attack.core.attack_config import AttackAdversarialConfig, AttackScoringConfig from pyrit.models import Parameter, SeedAttackGroup from pyrit.prompt_target import PromptTarget from pyrit.scenario.core.atomic_attack import AtomicAttack from pyrit.scenario.core.attack_technique import AttackTechnique -from pyrit.scenario.core.dataset_configuration import ( - DatasetAttackConfiguration, -) +from pyrit.scenario.core.dataset_configuration import DatasetAttackConfiguration from pyrit.scenario.core.scenario import Scenario from pyrit.scenario.core.scenario_context import ScenarioContext from pyrit.scenario.core.scenario_strategy import ScenarioStrategy @@ -127,7 +113,6 @@ def __init__( objective_scorer: TrueFalseScorer | None = None, adversarial_chat: PromptTarget | None = None, scenario_result_id: str | None = None, - include_baseline: bool | None = None, # Deprecated. Will be removed in 0.16.0. ) -> None: """ Initialize the ScamScenario. @@ -138,8 +123,6 @@ def __init__( adversarial_chat (PromptTarget | None): Chat target used to rephrase the objective into the role-play context (in single-turn strategies). scenario_result_id (str | None): Optional ID of an existing scenario result to resume. - include_baseline (bool | None): **Deprecated.** Will be removed in 0.16.0. Pass - ``include_baseline`` to ``initialize_async`` instead. """ if not objective_scorer: objective_scorer = self._get_default_objective_scorer() @@ -158,16 +141,6 @@ def __init__( scenario_result_id=scenario_result_id, ) - # Deprecated constructor-time baseline override. Will be removed in 0.16.0, along with - # the include_baseline kwarg above. - if include_baseline is not None: - print_deprecation_message( - old_item="Scam(include_baseline=...)", - new_item="Scam.initialize_async(include_baseline=...)", - removed_in="0.16.0", - ) - self._legacy_include_baseline = include_baseline - def _get_atomic_attack_from_strategy(self, *, strategy: str, seed_groups: list[SeedAttackGroup]) -> AtomicAttack: """ Translate the strategies into actual AtomicAttacks. diff --git a/pyrit/scenario/scenarios/benchmark/adversarial.py b/pyrit/scenario/scenarios/benchmark/adversarial.py index 9bfd8b056c..93480c9898 100644 --- a/pyrit/scenario/scenarios/benchmark/adversarial.py +++ b/pyrit/scenario/scenarios/benchmark/adversarial.py @@ -11,20 +11,12 @@ from pyrit.analytics import get_cached_results_for_technique from pyrit.common import apply_defaults -from pyrit.models import ( - AttackOutcome, - AttackResult, - ObjectiveTargetEvaluationIdentifier, - ScenarioResult, -) +from pyrit.models import AttackOutcome, AttackResult, ObjectiveTargetEvaluationIdentifier, ScenarioResult from pyrit.models.parameter import Parameter from pyrit.registry import AttackTechniqueRegistry, TargetRegistry from pyrit.registry.tag_query import TagQuery from pyrit.scenario.core.dataset_configuration import DatasetAttackConfiguration -from pyrit.scenario.core.matrix_atomic_attack_builder import ( - MatrixAtomicAttackBuilder, - resolve_technique_factories, -) +from pyrit.scenario.core.matrix_atomic_attack_builder import MatrixAtomicAttackBuilder, resolve_technique_factories from pyrit.scenario.core.scenario import BaselineAttackPolicy, Scenario if TYPE_CHECKING: diff --git a/pyrit/scenario/scenarios/foundry/__init__.py b/pyrit/scenario/scenarios/foundry/__init__.py index 2407b5674c..8939d59af8 100644 --- a/pyrit/scenario/scenarios/foundry/__init__.py +++ b/pyrit/scenario/scenarios/foundry/__init__.py @@ -3,11 +3,7 @@ """Foundry scenario classes.""" -from pyrit.scenario.scenarios.foundry.red_team_agent import ( - FoundryComposite, - FoundryStrategy, - RedTeamAgent, -) +from pyrit.scenario.scenarios.foundry.red_team_agent import FoundryComposite, FoundryStrategy, RedTeamAgent __all__ = [ "FoundryComposite", diff --git a/pyrit/scenario/scenarios/foundry/red_team_agent.py b/pyrit/scenario/scenarios/foundry/red_team_agent.py index ab9384202b..5faefa9255 100644 --- a/pyrit/scenario/scenarios/foundry/red_team_agent.py +++ b/pyrit/scenario/scenarios/foundry/red_team_agent.py @@ -16,19 +16,9 @@ from typing import TYPE_CHECKING, Any, TypeVar, cast from pyrit.common import REQUIRED_VALUE, apply_defaults -from pyrit.common.deprecation import print_deprecation_message # Deprecated. Will be removed in 0.16.0. from pyrit.datasets import TextJailBreak -from pyrit.executor.attack import ( - CrescendoAttack, - PromptSendingAttack, - RedTeamingAttack, - TreeOfAttacksWithPruningAttack, -) -from pyrit.executor.attack.core.attack_config import ( - AttackAdversarialConfig, - AttackConverterConfig, - AttackScoringConfig, -) +from pyrit.executor.attack import CrescendoAttack, PromptSendingAttack, RedTeamingAttack, TreeOfAttacksWithPruningAttack +from pyrit.executor.attack.core.attack_config import AttackAdversarialConfig, AttackConverterConfig, AttackScoringConfig from pyrit.models import SeedAttackGroup from pyrit.prompt_converter import ( AnsiAttackConverter, @@ -53,12 +43,8 @@ UrlConverter, ) from pyrit.prompt_converter.binary_converter import BinaryConverter -from pyrit.prompt_converter.token_smuggling.ascii_smuggler_converter import ( - AsciiSmugglerConverter, -) -from pyrit.prompt_normalizer.prompt_converter_configuration import ( - PromptConverterConfiguration, -) +from pyrit.prompt_converter.token_smuggling.ascii_smuggler_converter import AsciiSmugglerConverter +from pyrit.prompt_normalizer.prompt_converter_configuration import PromptConverterConfiguration from pyrit.prompt_target import PromptTarget from pyrit.scenario.core.atomic_attack import AtomicAttack from pyrit.scenario.core.attack_technique import AttackTechnique @@ -223,7 +209,6 @@ def __init__( adversarial_chat: PromptTarget | None = None, attack_scoring_config: AttackScoringConfig | None = None, scenario_result_id: str | None = None, - include_baseline: bool | None = None, # Deprecated. Will be removed in 0.16.0. ) -> None: """ Initialize a Foundry Scenario with the specified attack strategies. @@ -236,8 +221,6 @@ def __init__( including the objective scorer and auxiliary scorers. If not provided, creates a default configuration with a composite scorer using Azure Content Filter and SelfAsk Refusal scorers. scenario_result_id (str | None): Optional ID of an existing scenario result to resume. - include_baseline (bool | None): **Deprecated.** Will be removed in 0.16.0. Pass - ``include_baseline`` to ``initialize_async`` instead. Raises: ValueError: If attack_strategies is empty or contains unsupported strategies. @@ -264,16 +247,6 @@ def __init__( scenario_result_id=scenario_result_id, ) - # Deprecated constructor-time baseline override. Will be removed in 0.16.0, along with - # the include_baseline kwarg above. - if include_baseline is not None: - print_deprecation_message( - old_item="RedTeamAgent(include_baseline=...)", - new_item="RedTeamAgent.initialize_async(include_baseline=...)", - removed_in="0.16.0", - ) - self._legacy_include_baseline = include_baseline - self._scenario_composites: list[FoundryComposite] = [] @apply_defaults @@ -304,12 +277,15 @@ async def initialize_async( memory_labels (dict[str, str] | None): Labels to attach to all memory entries. include_baseline (bool | None): See ``Scenario.initialize_async``. """ - # 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). + # This override exists to widen the accepted strategy types (FoundryComposite is a + # dataclass, not a ScenarioStrategy enum member) and to expand composites up-front: + # _resolve_foundry_strategies populates self._scenario_composites (consumed by + # _build_atomic_attacks_async) and returns the flat concrete strategy list the base + # class tracks. + flat_strategies = self._resolve_foundry_strategies(scenario_strategies) await super().initialize_async( objective_target=objective_target, - scenario_strategies=scenario_strategies, + scenario_strategies=flat_strategies, dataset_config=dataset_config, max_concurrency=max_concurrency, max_retries=max_retries, @@ -317,7 +293,7 @@ async def initialize_async( include_baseline=include_baseline, ) - def _prepare_strategies( # type: ignore[ty:invalid-method-override] + def _resolve_foundry_strategies( self, strategies: "Sequence[FoundryStrategy | FoundryComposite | ScenarioCompositeStrategy] | None", ) -> list[ScenarioStrategy]: diff --git a/pyrit/scenario/scenarios/garak/__init__.py b/pyrit/scenario/scenarios/garak/__init__.py index 2b86927a4d..b394c8e5bf 100644 --- a/pyrit/scenario/scenarios/garak/__init__.py +++ b/pyrit/scenario/scenarios/garak/__init__.py @@ -5,10 +5,7 @@ from pyrit.scenario.scenarios.garak.doctor import Doctor, DoctorStrategy from pyrit.scenario.scenarios.garak.encoding import Encoding, EncodingStrategy -from pyrit.scenario.scenarios.garak.web_injection import ( - WebInjection, - WebInjectionStrategy, -) +from pyrit.scenario.scenarios.garak.web_injection import WebInjection, WebInjectionStrategy __all__ = [ "Doctor", diff --git a/pyrit/scenario/scenarios/garak/doctor.py b/pyrit/scenario/scenarios/garak/doctor.py index 116998fada..a0cbb7c01b 100644 --- a/pyrit/scenario/scenarios/garak/doctor.py +++ b/pyrit/scenario/scenarios/garak/doctor.py @@ -144,11 +144,10 @@ async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list """ Build the Doctor atomic attacks from the selected Policy Puppetry techniques. - Overrides the base extension point (rather than riding the base default cross-product - via ``_get_attack_technique_factories``) so the Doctor-specific techniques stay local - to this scenario and never enter the global registry. Delegates the technique × dataset - cross-product to ``MatrixAtomicAttackBuilder``. The base owns baseline emission, so this - passes ``include_baseline=False``. + Builds the Doctor-specific technique factories locally (so they never enter the global + registry) and delegates the technique × dataset cross-product to + ``MatrixAtomicAttackBuilder``. The base owns baseline emission, so this passes + ``include_baseline=False``. Args: context (ScenarioContext): The resolved runtime inputs for this run. diff --git a/pyrit/scenario/scenarios/garak/encoding.py b/pyrit/scenario/scenarios/garak/encoding.py index 2cf255753a..6300ddc3cc 100644 --- a/pyrit/scenario/scenarios/garak/encoding.py +++ b/pyrit/scenario/scenarios/garak/encoding.py @@ -6,11 +6,7 @@ from collections.abc import Sequence from pyrit.common import apply_defaults -from pyrit.common.deprecation import print_deprecation_message # Deprecated. Will be removed in 0.16.0. -from pyrit.executor.attack.core.attack_config import ( - AttackConverterConfig, - AttackScoringConfig, -) +from pyrit.executor.attack.core.attack_config import AttackConverterConfig, AttackScoringConfig from pyrit.executor.attack.single_turn.prompt_sending import PromptSendingAttack from pyrit.models import Seed, SeedAttackGroup, SeedObjective, SeedPrompt from pyrit.prompt_converter import ( @@ -29,15 +25,10 @@ from pyrit.prompt_converter.braille_converter import BrailleConverter from pyrit.prompt_converter.ecoji_converter import EcojiConverter from pyrit.prompt_converter.nato_converter import NatoConverter -from pyrit.prompt_normalizer.prompt_converter_configuration import ( - PromptConverterConfiguration, -) +from pyrit.prompt_normalizer.prompt_converter_configuration import PromptConverterConfiguration from pyrit.scenario.core.atomic_attack import AtomicAttack from pyrit.scenario.core.attack_technique import AttackTechnique -from pyrit.scenario.core.dataset_configuration import ( - CompoundDatasetAttackConfiguration, - DatasetAttackConfiguration, -) +from pyrit.scenario.core.dataset_configuration import CompoundDatasetAttackConfiguration, DatasetAttackConfiguration from pyrit.scenario.core.scenario import Scenario from pyrit.scenario.core.scenario_context import ScenarioContext from pyrit.scenario.core.scenario_strategy import ScenarioStrategy @@ -143,7 +134,6 @@ def __init__( objective_scorer: TrueFalseScorer | None = None, encoding_templates: Sequence[str] | None = None, scenario_result_id: str | None = None, - include_baseline: bool | None = None, # Deprecated. Will be removed in 0.16.0. ) -> None: """ Initialize the Encoding Scenario. @@ -155,8 +145,6 @@ def __init__( encoding_templates (Sequence[str] | None): Templates used to construct the decoding prompts. Defaults to AskToDecodeConverter.garak_templates. scenario_result_id (str | None): Optional ID of an existing scenario result to resume. - include_baseline (bool | None): **Deprecated.** Will be removed in 0.16.0. Pass - ``include_baseline`` to ``initialize_async`` instead. """ objective_scorer = objective_scorer or DecodingScorer(categories=["encoding_scenario"]) self._scorer_config = AttackScoringConfig(objective_scorer=objective_scorer) @@ -177,16 +165,6 @@ def __init__( scenario_result_id=scenario_result_id, ) - # Deprecated constructor-time baseline override. Will be removed in 0.16.0, along with - # the include_baseline kwarg above. - if include_baseline is not None: - print_deprecation_message( - old_item="Encoding(include_baseline=...)", - new_item="Encoding.initialize_async(include_baseline=...)", - removed_in="0.16.0", - ) - self._legacy_include_baseline = include_baseline - async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list[AtomicAttack]: """ Build the encoding atomic attacks for this run. diff --git a/pyrit/scenario/scenarios/garak/web_injection.py b/pyrit/scenario/scenarios/garak/web_injection.py index 7de7649c43..33c9c41bb1 100644 --- a/pyrit/scenario/scenarios/garak/web_injection.py +++ b/pyrit/scenario/scenarios/garak/web_injection.py @@ -1,10 +1,11 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT license. +from __future__ import annotations import logging import random -from typing import ClassVar +from typing import TYPE_CHECKING, ClassVar from pyrit.common import apply_defaults from pyrit.executor.attack.core.attack_config import AttackScoringConfig @@ -16,14 +17,13 @@ from pyrit.scenario.core.dataset_configuration import DatasetConfiguration from pyrit.scenario.core.scenario import BaselineAttackPolicy, Scenario from pyrit.scenario.core.scenario_strategy import ScenarioStrategy -from pyrit.score import ( - TrueFalseCompositeScorer, - TrueFalseScoreAggregator, - TrueFalseScorer, -) +from pyrit.score import TrueFalseCompositeScorer, TrueFalseScoreAggregator, TrueFalseScorer from pyrit.score.true_false.regex.markdown_injection import MarkdownInjectionScorer from pyrit.score.true_false.regex.xss_output_scorer import XSSOutputScorer +if TYPE_CHECKING: + from pyrit.scenario.core.scenario_context import ScenarioContext + logger = logging.getLogger(__name__) @@ -498,27 +498,25 @@ def _scoring_config_for_strategy(self, strategy: WebInjectionStrategy) -> Attack return self._xss_scoring_config return self._exfil_scoring_config - async def _get_atomic_attacks_async(self) -> list[AtomicAttack]: + async def _resolve_seed_groups_by_dataset_async(self) -> dict[str, list[SeedAttackGroup]]: """ - Build one AtomicAttack per selected strategy, plus an optional baseline. + Generate the injection prompts and wrap them into seed groups, keyed by strategy. + + WebInjection synthesizes its seeds (rather than resolving them from a + ``DatasetAttackConfiguration``): each strategy renders its own objective and prompt + set from the raw garak datasets. Resolving them here means the base owns the single + seed sample used for both the atomic attacks and the central baseline. Returns: - list[AtomicAttack]: The atomic attacks for this scenario. + dict[str, list[SeedAttackGroup]]: Seed groups keyed by strategy value. Raises: - ValueError: If the scenario is not initialized or no prompts were generated. + ValueError: If no prompts were generated for any selected strategy. """ - if self._objective_target is None: - raise ValueError( - "Scenario not properly initialized. Call await scenario.initialize_async() before running." - ) - dataset_values = self._load_dataset_values() rng = random.Random(self._random_seed) - atomic_attacks: list[AtomicAttack] = [] - all_seed_groups: list[SeedAttackGroup] = [] - + seed_groups_by_strategy: dict[str, list[SeedAttackGroup]] = {} for strategy in self._scenario_strategies: objective, prompts = self._build_prompts_for_strategy( strategy=strategy, dataset_values=dataset_values, rng=rng @@ -528,10 +526,38 @@ async def _get_atomic_attacks_async(self) -> list[AtomicAttack]: continue seed_groups = self._build_seed_groups(objective=objective, prompts=prompts) - all_seed_groups.extend(seed_groups) + if seed_groups: + seed_groups_by_strategy[strategy.value] = seed_groups + if not seed_groups_by_strategy: + raise ValueError( + "WebInjection scenario produced no prompts. Ensure the garak web-injection datasets " + "(garak_example_domains_xss, garak_markdown_js, garak_web_html_js, " + "garak_xss_normal_instructions) are loaded into CentralMemory before running." + ) + + return seed_groups_by_strategy + + async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list[AtomicAttack]: + """ + Build one AtomicAttack per selected strategy from the resolved seed groups. + + The base owns baseline emission (from ``context.seed_groups``), so this never + prepends one itself. + + Args: + context (ScenarioContext): The resolved runtime inputs for this run. + + Returns: + list[AtomicAttack]: The atomic attacks for this scenario. + """ + strategies_by_value = {strategy.value: strategy for strategy in context.scenario_strategies} + + atomic_attacks: list[AtomicAttack] = [] + for name, seed_groups in context.seed_groups_by_dataset.items(): + strategy = strategies_by_value[name] attack = PromptSendingAttack( - objective_target=self._objective_target, + objective_target=context.objective_target, attack_scoring_config=self._scoring_config_for_strategy(strategy), ) atomic_attacks.append( @@ -539,18 +565,8 @@ async def _get_atomic_attacks_async(self) -> list[AtomicAttack]: atomic_attack_name=strategy.value, attack_technique=AttackTechnique(attack=attack), seed_groups=seed_groups, - memory_labels=self._memory_labels, + memory_labels=context.memory_labels, ) ) - if not atomic_attacks: - raise ValueError( - "WebInjection scenario produced no prompts. Ensure the garak web-injection datasets " - "(garak_example_domains_xss, garak_markdown_js, garak_web_html_js, " - "garak_xss_normal_instructions) are loaded into CentralMemory before running." - ) - - if self._include_baseline and all_seed_groups: - atomic_attacks.insert(0, self._build_baseline_atomic_attack(seed_groups=all_seed_groups)) - return atomic_attacks diff --git a/tests/unit/scenario/airt/test_cyber.py b/tests/unit/scenario/airt/test_cyber.py index 87be865481..8d15e0c9e2 100644 --- a/tests/unit/scenario/airt/test_cyber.py +++ b/tests/unit/scenario/airt/test_cyber.py @@ -11,15 +11,10 @@ from pyrit.models import ComponentIdentifier, SeedAttackGroup, SeedObjective, SeedPrompt from pyrit.prompt_target import PromptTarget from pyrit.registry.components.attack_technique_registry import AttackTechniqueRegistry -from pyrit.scenario.core.dataset_configuration import ( - DatasetAttackConfiguration, - DatasetConfiguration, -) +from pyrit.scenario.core.dataset_configuration import DatasetAttackConfiguration, DatasetConfiguration from pyrit.scenario.scenarios.airt.cyber import Cyber from pyrit.score import TrueFalseScorer -from pyrit.setup.initializers.components.scenario_techniques import ( - build_scenario_technique_factories, -) +from pyrit.setup.initializers.components.scenario_techniques import build_scenario_technique_factories # --------------------------------------------------------------------------- # Helpers @@ -258,7 +253,7 @@ async def _init_and_get_attacks( if strategies: init_kwargs["scenario_strategies"] = strategies await scenario.initialize_async(**init_kwargs) - return await scenario._get_atomic_attacks_async() + return scenario._atomic_attacks async def test_all_strategy_produces_red_teaming(self, mock_objective_target, mock_objective_scorer): attacks = await self._init_and_get_attacks( @@ -321,7 +316,7 @@ async def test_attacks_include_seed_groups(self, mock_objective_target, mock_obj async def test_raises_when_not_initialized(self, mock_objective_scorer): scenario = Cyber(objective_scorer=mock_objective_scorer) with pytest.raises(ValueError, match="Scenario not properly initialized"): - await scenario._get_atomic_attacks_async() + scenario._build_scenario_context(seed_groups_by_dataset={}) # =========================================================================== @@ -349,17 +344,17 @@ class TestCyberRegistryIntegration: """Tests for attack technique registry wiring via Cyber scenario.""" def test_cyber_factories_include_red_teaming(self, mock_objective_scorer): - scenario = Cyber(objective_scorer=mock_objective_scorer) - factories = scenario._get_attack_technique_factories() - # Cyber filters the registry to red_teaming; the PromptSendingAttack baseline + registry = AttackTechniqueRegistry.get_registry_singleton() + factories = registry.get_factories_or_raise() + # Cyber selects red_teaming from the registry; the PromptSendingAttack baseline # is contributed at runtime by BaselineAttackPolicy.Enabled, not by this dict. assert "red_teaming" in factories assert factories["red_teaming"].attack_class is RedTeamingAttack def test_red_teaming_factory_has_adversarial_config(self, mock_objective_scorer): """red_teaming factory advertises uses_adversarial (config resolved lazily at create()).""" - scenario = Cyber(objective_scorer=mock_objective_scorer) - factories = scenario._get_attack_technique_factories() + registry = AttackTechniqueRegistry.get_registry_singleton() + factories = registry.get_factories_or_raise() assert factories["red_teaming"].uses_adversarial is True assert factories["red_teaming"]._adversarial_chat is None diff --git a/tests/unit/scenario/airt/test_jailbreak.py b/tests/unit/scenario/airt/test_jailbreak.py index 6376641988..73234bb168 100644 --- a/tests/unit/scenario/airt/test_jailbreak.py +++ b/tests/unit/scenario/airt/test_jailbreak.py @@ -294,7 +294,7 @@ async def test_attack_generation_for_simple( await scenario.initialize_async( objective_target=mock_objective_target, scenario_strategies=[simple_jailbreak_strategy] ) - atomic_attacks = await scenario._get_atomic_attacks_async() + atomic_attacks = scenario._atomic_attacks for run in atomic_attacks: assert isinstance(run.attack_technique.attack, PromptSendingAttack) @@ -315,7 +315,7 @@ async def test_attack_generation_for_complex( scenario_strategies=[complex_jailbreak_strategy], include_baseline=False, ) - atomic_attacks = await scenario._get_atomic_attacks_async() + atomic_attacks = scenario._atomic_attacks for run in atomic_attacks: assert isinstance( run.attack_technique.attack, (RolePlayAttack, ManyShotJailbreakAttack, SkeletonKeyAttack) @@ -338,7 +338,7 @@ async def test_attack_generation_for_manyshot( scenario_strategies=[manyshot_jailbreak_strategy], include_baseline=False, ) - atomic_attacks = await scenario._get_atomic_attacks_async() + atomic_attacks = scenario._atomic_attacks for run in atomic_attacks: assert isinstance(run.attack_technique.attack, ManyShotJailbreakAttack) @@ -359,7 +359,7 @@ async def test_attack_generation_for_promptsending( scenario_strategies=[promptsending_jailbreak_strategy], include_baseline=False, ) - atomic_attacks = await scenario._get_atomic_attacks_async() + atomic_attacks = scenario._atomic_attacks for run in atomic_attacks: assert isinstance(run.attack_technique.attack, PromptSendingAttack) @@ -380,7 +380,7 @@ async def test_attack_generation_for_skeleton( scenario_strategies=[skeleton_jailbreak_attack], include_baseline=False, ) - atomic_attacks = await scenario._get_atomic_attacks_async() + atomic_attacks = scenario._atomic_attacks for run in atomic_attacks: assert isinstance(run.attack_technique.attack, SkeletonKeyAttack) @@ -401,7 +401,7 @@ async def test_attack_generation_for_roleplay( scenario_strategies=[roleplay_jailbreak_strategy], include_baseline=False, ) - atomic_attacks = await scenario._get_atomic_attacks_async() + atomic_attacks = scenario._atomic_attacks for run in atomic_attacks: assert isinstance(run.attack_technique.attack, RolePlayAttack) @@ -422,7 +422,7 @@ async def test_attack_runs_include_objectives( scenario = Jailbreak(objective_scorer=mock_objective_scorer, num_templates=2) await scenario.initialize_async(objective_target=mock_objective_target) - atomic_attacks = await scenario._get_atomic_attacks_async() + atomic_attacks = scenario._atomic_attacks assert len(atomic_attacks) > 0 for run in atomic_attacks: @@ -471,7 +471,7 @@ async def test_custom_num_attempts( ): base_scenario = Jailbreak(objective_scorer=mock_objective_scorer, num_templates=2) await base_scenario.initialize_async(objective_target=mock_objective_target, include_baseline=False) - atomic_attacks_1 = await base_scenario._get_atomic_attacks_async() + atomic_attacks_1 = base_scenario._atomic_attacks mult_scenario = Jailbreak( objective_scorer=mock_objective_scorer, @@ -479,7 +479,7 @@ async def test_custom_num_attempts( num_attempts=mock_random_num_attempts, ) await mult_scenario.initialize_async(objective_target=mock_objective_target, include_baseline=False) - atomic_attacks_n = await mult_scenario._get_atomic_attacks_async() + atomic_attacks_n = mult_scenario._atomic_attacks assert len(atomic_attacks_1) * mock_random_num_attempts == len(atomic_attacks_n) @@ -617,7 +617,7 @@ async def test_roleplay_attacks_share_adversarial_target( scenario_strategies=[roleplay_jailbreak_strategy], include_baseline=False, ) - atomic_attacks = await scenario._get_atomic_attacks_async() + atomic_attacks = scenario._atomic_attacks assert len(atomic_attacks) >= 2 # All role-play attacks should share the same adversarial target diff --git a/tests/unit/scenario/airt/test_leakage.py b/tests/unit/scenario/airt/test_leakage.py index c6933183b3..bc9302f5a4 100644 --- a/tests/unit/scenario/airt/test_leakage.py +++ b/tests/unit/scenario/airt/test_leakage.py @@ -136,7 +136,7 @@ async def test_attack_generation_for_all(self, mock_objective_target, mock_objec """Test that _get_atomic_attacks_async returns atomic attacks.""" scenario = Leakage(objective_scorer=mock_objective_scorer) await scenario.initialize_async(objective_target=mock_objective_target, dataset_config=mock_dataset_config) - atomic_attacks = await scenario._get_atomic_attacks_async() + atomic_attacks = scenario._atomic_attacks assert len(atomic_attacks) > 0 assert all(run.attack_technique is not None for run in atomic_attacks) @@ -147,7 +147,7 @@ async def test_attack_runs_include_objectives( """Test that attack runs include objectives for each seed prompt.""" scenario = Leakage(objective_scorer=mock_objective_scorer) await scenario.initialize_async(objective_target=mock_objective_target, dataset_config=mock_dataset_config) - atomic_attacks = await scenario._get_atomic_attacks_async() + atomic_attacks = scenario._atomic_attacks for run in atomic_attacks: assert len(run.objectives) > 0 diff --git a/tests/unit/scenario/airt/test_psychosocial.py b/tests/unit/scenario/airt/test_psychosocial.py index e62fbeba4b..7f91a37152 100644 --- a/tests/unit/scenario/airt/test_psychosocial.py +++ b/tests/unit/scenario/airt/test_psychosocial.py @@ -10,10 +10,7 @@ from pyrit.common.path import DATASETS_PATH from pyrit.models import ComponentIdentifier, SeedAttackGroup, SeedDataset, SeedGroup, SeedObjective from pyrit.prompt_target import OpenAIChatTarget, PromptTarget -from pyrit.scenario.airt import ( # type: ignore[ty:unresolved-import] - Psychosocial, - PsychosocialStrategy, -) +from pyrit.scenario.airt import Psychosocial, PsychosocialStrategy # type: ignore[ty:unresolved-import] from pyrit.scenario.scenarios.airt.psychosocial import SubharmConfig from pyrit.score import FloatScaleThresholdScorer @@ -196,7 +193,7 @@ async def test_attack_generation_for_all( scenario = Psychosocial(objective_scorer=mock_objective_scorer) await scenario.initialize_async(objective_target=mock_objective_target, dataset_config=mock_dataset_config) - atomic_attacks = await scenario._get_atomic_attacks_async() + atomic_attacks = scenario._atomic_attacks assert len(atomic_attacks) > 0 assert all(run.attack_technique is not None for run in atomic_attacks) @@ -221,7 +218,7 @@ async def test_attack_runs_include_objectives_async( ) await scenario.initialize_async(objective_target=mock_objective_target, dataset_config=mock_dataset_config) - atomic_attacks = await scenario._get_atomic_attacks_async() + atomic_attacks = scenario._atomic_attacks for run in atomic_attacks: assert len(run.objectives) > 0 @@ -246,7 +243,7 @@ async def test_get_atomic_attacks_async_returns_attacks( ) await scenario.initialize_async(objective_target=mock_objective_target, dataset_config=mock_dataset_config) - atomic_attacks = await scenario._get_atomic_attacks_async() + atomic_attacks = scenario._atomic_attacks assert len(atomic_attacks) > 0 assert all(run.attack_technique is not None for run in atomic_attacks) diff --git a/tests/unit/scenario/airt/test_rapid_response.py b/tests/unit/scenario/airt/test_rapid_response.py index 2ef1247540..a2fe8dc118 100644 --- a/tests/unit/scenario/airt/test_rapid_response.py +++ b/tests/unit/scenario/airt/test_rapid_response.py @@ -21,16 +21,10 @@ from pyrit.registry import TargetRegistry from pyrit.registry.components.attack_technique_registry import AttackTechniqueRegistry from pyrit.scenario.core.attack_technique_factory import AttackTechniqueFactory -from pyrit.scenario.core.dataset_configuration import ( - CompoundDatasetAttackConfiguration, -) -from pyrit.scenario.scenarios.airt.rapid_response import ( - RapidResponse, -) +from pyrit.scenario.core.dataset_configuration import CompoundDatasetAttackConfiguration +from pyrit.scenario.scenarios.airt.rapid_response import RapidResponse from pyrit.score import TrueFalseScorer -from pyrit.setup.initializers.components.scenario_techniques import ( - build_scenario_technique_factories, -) +from pyrit.setup.initializers.components.scenario_techniques import build_scenario_technique_factories # --------------------------------------------------------------------------- # Synthetic many-shot examples — prevents reading the real JSON during tests @@ -295,7 +289,7 @@ async def _init_and_get_attacks( if strategies: init_kwargs["scenario_strategies"] = strategies await scenario.initialize_async(**init_kwargs) - return await scenario._get_atomic_attacks_async() + return scenario._atomic_attacks async def test_default_strategy_produces_role_play_and_many_shot( self, mock_objective_target, mock_objective_scorer @@ -397,7 +391,6 @@ def _spy_create(self, **kwargs): scenario_strategies=[role_play], strategy_converters={role_play.value: [converter]}, ) - await scenario._get_atomic_attacks_async() # ROLE_PLAY was selected with a converter modifier, so every resulting factory.create # call must receive the extra request converter. @@ -457,7 +450,7 @@ async def test_raises_when_not_initialized(self, mock_objective_scorer): objective_scorer=mock_objective_scorer, ) with pytest.raises(ValueError, match="Scenario not properly initialized"): - await scenario._get_atomic_attacks_async() + scenario._build_scenario_context(seed_groups_by_dataset={}) async def test_unknown_technique_skipped_with_warning(self, mock_objective_target, mock_objective_scorer): """If a technique name has no factory, it's skipped (not an error).""" @@ -493,7 +486,7 @@ async def test_unknown_technique_skipped_with_warning(self, mock_objective_targe scenario_strategies=[_strategy_class().ALL], include_baseline=False, ) - attacks = await scenario._get_atomic_attacks_async() + attacks = scenario._atomic_attacks # Only prompt_sending should have produced attacks assert len(attacks) == 1 assert isinstance(attacks[0].attack_technique.attack, PromptSendingAttack) @@ -519,8 +512,8 @@ class TestCoreTechniques: """Tests for shared AttackTechniqueFactory builders in scenario_techniques.py.""" def test_instance_returns_all_factories(self, mock_objective_scorer): - scenario = RapidResponse(objective_scorer=mock_objective_scorer) - factories = scenario._get_attack_technique_factories() + registry = AttackTechniqueRegistry.get_registry_singleton() + factories = registry.get_factories() assert {"role_play", "many_shot", "tap"} <= set(factories.keys()) assert factories["role_play"].attack_class is RolePlayAttack assert factories["many_shot"].attack_class is ManyShotJailbreakAttack @@ -532,8 +525,8 @@ def test_factories_use_default_adversarial_when_none(self, mock_objective_scorer The default adversarial target is resolved lazily inside ``create()``; it is not baked into the factory at construction time. """ - scenario = RapidResponse(objective_scorer=mock_objective_scorer) - factories = scenario._get_attack_technique_factories() + registry = AttackTechniqueRegistry.get_registry_singleton() + factories = registry.get_factories() assert factories["role_play"].uses_adversarial is True assert factories["tap"].uses_adversarial is True assert factories["role_play"]._adversarial_chat is None @@ -541,8 +534,8 @@ def test_factories_use_default_adversarial_when_none(self, mock_objective_scorer def test_factories_always_use_default_adversarial(self, mock_objective_scorer): """Factories defer adversarial wiring to create()-time lazy resolution.""" - scenario = RapidResponse(objective_scorer=mock_objective_scorer) - factories = scenario._get_attack_technique_factories() + registry = AttackTechniqueRegistry.get_registry_singleton() + factories = registry.get_factories() assert factories["role_play"]._adversarial_chat is None assert factories["tap"]._adversarial_chat is None @@ -587,12 +580,6 @@ def test_get_factories_returns_dict(self): assert {"role_play", "many_shot", "tap"} <= set(factories.keys()) assert factories["role_play"].attack_class is RolePlayAttack - def test_scenario_base_class_reads_from_registry(self, mock_objective_scorer): - """Scenario._get_attack_technique_factories() reads from the registry.""" - scenario = RapidResponse(objective_scorer=mock_objective_scorer) - factories = scenario._get_attack_technique_factories() - assert {"role_play", "many_shot", "tap"} <= set(factories.keys()) - def test_tags_assigned_correctly(self): registry = AttackTechniqueRegistry.get_registry_singleton() single_turn = {e.name for e in registry.instances.get_by_tag(tag="single_turn")} diff --git a/tests/unit/scenario/airt/test_scam.py b/tests/unit/scenario/airt/test_scam.py index 05b4760e8c..f0f71c588f 100644 --- a/tests/unit/scenario/airt/test_scam.py +++ b/tests/unit/scenario/airt/test_scam.py @@ -9,11 +9,7 @@ import pytest from pyrit.common.path import DATASETS_PATH -from pyrit.executor.attack import ( - ContextComplianceAttack, - RedTeamingAttack, - RolePlayAttack, -) +from pyrit.executor.attack import ContextComplianceAttack, RedTeamingAttack, RolePlayAttack from pyrit.executor.attack.core.attack_config import AttackScoringConfig from pyrit.models import ComponentIdentifier, SeedAttackGroup, SeedDataset, SeedObjective from pyrit.prompt_target import OpenAIChatTarget, PromptTarget @@ -233,7 +229,7 @@ async def test_attack_generation_for_all( scenario = Scam(objective_scorer=mock_objective_scorer) await scenario.initialize_async(objective_target=mock_objective_target, dataset_config=mock_dataset_config) - atomic_attacks = await scenario._get_atomic_attacks_async() + atomic_attacks = scenario._atomic_attacks assert len(atomic_attacks) > 0 assert all(run.attack_technique is not None for run in atomic_attacks) @@ -257,7 +253,7 @@ async def test_attack_generation_for_singleturn_async( dataset_config=mock_dataset_config, include_baseline=False, ) - atomic_attacks = await scenario._get_atomic_attacks_async() + atomic_attacks = scenario._atomic_attacks for run in atomic_attacks: assert isinstance(run.attack_technique.attack, (ContextComplianceAttack, RolePlayAttack)) @@ -276,7 +272,7 @@ async def test_attack_generation_for_multiturn_async( dataset_config=mock_dataset_config, include_baseline=False, ) - atomic_attacks = await scenario._get_atomic_attacks_async() + atomic_attacks = scenario._atomic_attacks for run in atomic_attacks: assert isinstance(run.attack_technique.attack, RedTeamingAttack) @@ -295,7 +291,7 @@ async def test_attack_runs_include_objectives_async( ) await scenario.initialize_async(objective_target=mock_objective_target, dataset_config=mock_dataset_config) - atomic_attacks = await scenario._get_atomic_attacks_async() + atomic_attacks = scenario._atomic_attacks for run in atomic_attacks: assert len(run.objectives) == len(mock_memory_seeds) @@ -315,7 +311,7 @@ async def test_get_atomic_attacks_async_returns_attacks( ) await scenario.initialize_async(objective_target=mock_objective_target, dataset_config=mock_dataset_config) - atomic_attacks = await scenario._get_atomic_attacks_async() + atomic_attacks = scenario._atomic_attacks assert len(atomic_attacks) > 0 assert all(run.attack_technique is not None for run in atomic_attacks) @@ -343,7 +339,7 @@ async def test_max_turns_default_used_when_unset_async( dataset_config=mock_dataset_config, include_baseline=False, ) - atomic_attacks = await scenario._get_atomic_attacks_async() + atomic_attacks = scenario._atomic_attacks for run in atomic_attacks: assert isinstance(run.attack_technique.attack, RedTeamingAttack) @@ -362,7 +358,7 @@ async def test_max_turns_override_flows_into_attack_async( dataset_config=mock_dataset_config, include_baseline=False, ) - atomic_attacks = await scenario._get_atomic_attacks_async() + atomic_attacks = scenario._atomic_attacks for run in atomic_attacks: assert run.attack_technique.attack._max_turns == 10 diff --git a/tests/unit/scenario/benchmark/test_adversarial.py b/tests/unit/scenario/benchmark/test_adversarial.py index 66f6a1c5ea..6c7f60d53a 100644 --- a/tests/unit/scenario/benchmark/test_adversarial.py +++ b/tests/unit/scenario/benchmark/test_adversarial.py @@ -8,7 +8,7 @@ ``supported_parameters``. Targets are user-supplied registry names that resolve to ``PromptTarget`` instances via ``TargetRegistry``. The ``(technique × target × dataset)`` cross-product is built lazily inside -``_get_atomic_attacks_async`` using factory.create() with an +``_build_atomic_attacks_async`` using factory.create() with an adversarial config override; no global ``AttackTechniqueRegistry`` state is mutated. @@ -19,7 +19,7 @@ source ``light`` tag (excludes ``tap`` / ``crescendo_simulated``). * ``supported_parameters`` declares ``adversarial_targets: list[str]``. * ``_resolve_adversarial_targets`` raises with available names on typos. -* ``_get_atomic_attacks_async`` produces ``N × M × D`` atomic attacks +* ``_build_atomic_attacks_async`` produces ``N × M × D`` atomic attacks with the expected ``atomic_attack_name`` and ``display_group``. * ``_collect_cached_completion_pairs`` delegates to ``pyrit.analytics.get_cached_results_for_technique`` per unique @@ -52,10 +52,7 @@ from pyrit.scenario.core import BaselineAttackPolicy from pyrit.scenario.core.attack_technique_factory import AttackTechniqueFactory from pyrit.scenario.core.scenario import Scenario -from pyrit.scenario.scenarios.benchmark.adversarial import ( - AdversarialBenchmark, - _build_benchmark_strategy, -) +from pyrit.scenario.scenarios.benchmark.adversarial import AdversarialBenchmark, _build_benchmark_strategy from pyrit.score import TrueFalseScorer from pyrit.setup.initializers.components.scenario_techniques import build_scenario_technique_factories @@ -141,6 +138,14 @@ def _register_mock_factory(*, name: str, tags: list[str] | None = None, seed_tec return factory +async def _build_atomic_attacks(bench: AdversarialBenchmark) -> list: + """Drive the post-``initialize_async`` build path: resolve seeds, snapshot the + context, then build atomic attacks — the same sequence ``initialize_async`` runs.""" + seed_groups_by_dataset = await bench._resolve_seed_groups_by_dataset_async() + context = bench._build_scenario_context(seed_groups_by_dataset=seed_groups_by_dataset) + return await bench._build_atomic_attacks_async(context=context) + + # --------------------------------------------------------------------------- # Class metadata # --------------------------------------------------------------------------- @@ -354,24 +359,24 @@ def test_preserves_caller_order(self): # --------------------------------------------------------------------------- -# _get_atomic_attacks_async — validation and cross-product +# _build_atomic_attacks_async — validation and cross-product # --------------------------------------------------------------------------- @pytest.mark.usefixtures("patch_central_database") class TestGetAtomicAttacksValidation: - """Tests for validation errors raised by ``_get_atomic_attacks_async``.""" + """Tests for validation errors raised by ``_build_atomic_attacks_async``.""" def _make_bench(self) -> AdversarialBenchmark: return AdversarialBenchmark(objective_scorer=MagicMock(spec=TrueFalseScorer)) async def test_uninitialized_scenario_raises(self): - """Calling ``_get_atomic_attacks_async`` before ``initialize_async`` raises a clear error.""" + """Building a context before ``initialize_async`` raises a clear error.""" bench = self._make_bench() bench._objective_target = None with pytest.raises(ValueError, match="not properly initialized"): - await bench._get_atomic_attacks_async() + bench._build_scenario_context(seed_groups_by_dataset={}) async def test_missing_adversarial_targets_raises_actionable_error(self): """Empty/missing ``adversarial_targets`` raises a message pointing at CLI / .pyrit_conf / list-targets.""" @@ -379,8 +384,9 @@ async def test_missing_adversarial_targets_raises_actionable_error(self): bench._objective_target = MagicMock(spec=PromptTarget) bench.params = {} + context = bench._build_scenario_context(seed_groups_by_dataset={}) with pytest.raises(ValueError) as exc_info: - await bench._get_atomic_attacks_async() + await bench._build_atomic_attacks_async(context=context) message = str(exc_info.value) assert "--adversarial-targets" in message @@ -392,8 +398,9 @@ async def test_empty_adversarial_targets_list_raises(self): bench._objective_target = MagicMock(spec=PromptTarget) bench.params = {"adversarial_targets": []} + context = bench._build_scenario_context(seed_groups_by_dataset={}) with pytest.raises(ValueError, match="at least one adversarial chat target"): - await bench._get_atomic_attacks_async() + await bench._build_atomic_attacks_async(context=context) async def test_unknown_target_name_raises_listing_available(self): _register_adversarial_target(name="adv_a") @@ -401,8 +408,9 @@ async def test_unknown_target_name_raises_listing_available(self): bench._objective_target = MagicMock(spec=PromptTarget) bench.params = {"adversarial_targets": ["missing"]} + context = bench._build_scenario_context(seed_groups_by_dataset={}) with pytest.raises(ValueError) as exc_info: - await bench._get_atomic_attacks_async() + await bench._build_atomic_attacks_async(context=context) message = str(exc_info.value) assert "missing" in message @@ -411,7 +419,7 @@ async def test_unknown_target_name_raises_listing_available(self): @pytest.mark.usefixtures("patch_central_database") class TestGetAtomicAttacksCrossProduct: - """Tests for the (technique × target × dataset) cross-product produced by ``_get_atomic_attacks_async``.""" + """Tests for the (technique × target × dataset) cross-product produced by ``_build_atomic_attacks_async``.""" def _make_bench_with_targets(self, *, target_names: list[str]) -> AdversarialBenchmark: for name in target_names: @@ -439,20 +447,20 @@ def _make_bench_with_targets(self, *, target_names: list[str]) -> AdversarialBen async def test_cross_product_count_matches_n_techniques_m_targets_d_datasets(self): """1 technique × 2 targets × 1 dataset = 2 atomic attacks.""" bench = self._make_bench_with_targets(target_names=["adv_a", "adv_b"]) - result = await bench._get_atomic_attacks_async() + result = await _build_atomic_attacks(bench) assert len(result) == 2 async def test_atomic_attack_name_format_is_technique__target_dataset(self): """Name format: ``{technique}__{target}_{dataset}`` (preserves VERSION=2 cache key shape).""" bench = self._make_bench_with_targets(target_names=["adv_a"]) - result = await bench._get_atomic_attacks_async() + result = await _build_atomic_attacks(bench) names = [a.atomic_attack_name for a in result] assert names == ["red_teaming__adv_a_harmbench"] async def test_display_group_equals_target_registry_name(self): """``display_group`` is the raw target registry name — no string parsing.""" bench = self._make_bench_with_targets(target_names=["adv_a", "adv_b"]) - result = await bench._get_atomic_attacks_async() + result = await _build_atomic_attacks(bench) display_groups = sorted({a.display_group for a in result}) assert display_groups == ["adv_a", "adv_b"] @@ -481,7 +489,7 @@ async def test_display_group_uses_registry_name_not_target_model_name(self): bench._dataset_config = MagicMock() bench._dataset_config.get_attack_groups_by_dataset_async = AsyncMock(return_value={"harmbench": [seed_group]}) - result = await bench._get_atomic_attacks_async() + result = await _build_atomic_attacks(bench) assert len(result) == 1 atomic = result[0] @@ -495,7 +503,7 @@ async def test_factory_create_called_per_target_with_adversarial_chat(self): bench = self._make_bench_with_targets(target_names=["adv_a", "adv_b"]) factory = AttackTechniqueRegistry.get_registry_singleton().get_factories_or_raise()["red_teaming"] - await bench._get_atomic_attacks_async() + await _build_atomic_attacks(bench) # 1 factory × 2 targets × 1 dataset = 2 create calls assert factory.create.call_count == 2 @@ -735,13 +743,13 @@ def test_identifier_construction_failure_falls_back_to_empty(self): # --------------------------------------------------------------------------- -# skip_cached end-to-end through _get_atomic_attacks_async +# skip_cached end-to-end through _build_atomic_attacks_async # --------------------------------------------------------------------------- @pytest.mark.usefixtures("patch_central_database") class TestSkipCachedFilter: - """End-to-end tests for the ``skip_cached`` filter applied in ``_get_atomic_attacks_async``.""" + """End-to-end tests for the ``skip_cached`` filter applied in ``_build_atomic_attacks_async``.""" _ANALYTICS_PATH = "pyrit.scenario.scenarios.benchmark.adversarial.get_cached_results_for_technique" _IDENTIFIER_PATH = "pyrit.scenario.scenarios.benchmark.adversarial.ObjectiveTargetEvaluationIdentifier" @@ -779,7 +787,7 @@ async def test_use_cached_false_returns_all_candidates_without_analytics_call(se bench = self._make_bench(use_cached=False) with patch(self._ANALYTICS_PATH) as analytics_mock: - result = await bench._get_atomic_attacks_async() + result = await _build_atomic_attacks(bench) assert len(result) == 1 analytics_mock.assert_not_called() @@ -803,7 +811,7 @@ async def test_use_cached_true_filters_matching_candidates(self): ], ), ): - result = await bench._get_atomic_attacks_async() + result = await _build_atomic_attacks(bench) assert result == [] @@ -814,7 +822,7 @@ async def test_use_cached_true_keeps_unmatched_candidates(self): self._patch_identifier(), patch(self._ANALYTICS_PATH, return_value=[]), ): - result = await bench._get_atomic_attacks_async() + result = await _build_atomic_attacks(bench) assert len(result) == 1 @@ -834,7 +842,7 @@ async def test_use_cached_true_populates_precomputed_maps_for_skipped(self): ), patch(self._ANALYTICS_PATH, return_value=[cached_attack]), ): - result = await bench._get_atomic_attacks_async() + result = await _build_atomic_attacks(bench) assert result == [] assert bench._precomputed_cached_results == {"red_teaming__adv_a_harmbench": [cached_attack]} @@ -860,7 +868,7 @@ async def test_use_cached_true_filters_results_by_parent_collection(self): ), patch(self._ANALYTICS_PATH, return_value=[matching, wrong_parent]), ): - await bench._get_atomic_attacks_async() + await _build_atomic_attacks(bench) assert bench._precomputed_cached_results == {"red_teaming__adv_a_harmbench": [matching]} @@ -1141,7 +1149,7 @@ async def test_precomputed_results_injected_into_attack_results(self): result_y = MagicMock(spec=AttackResult) result_z = MagicMock(spec=AttackResult) - # Simulate what _get_atomic_attacks_async populated for the two skipped attacks + # Simulate what _build_atomic_attacks_async populated for the two skipped attacks bench._precomputed_cached_results = { "technique_a__adv_target_harmbench": [result_x], "technique_b__adv_target_harmbench": [result_y], diff --git a/tests/unit/scenario/core/test_attack_technique_factory.py b/tests/unit/scenario/core/test_attack_technique_factory.py index 9cdb6cdfef..3b5fbd08ea 100644 --- a/tests/unit/scenario/core/test_attack_technique_factory.py +++ b/tests/unit/scenario/core/test_attack_technique_factory.py @@ -7,11 +7,7 @@ import pytest -from pyrit.executor.attack.core.attack_config import ( - AttackAdversarialConfig, - AttackConverterConfig, - AttackScoringConfig, -) +from pyrit.executor.attack.core.attack_config import AttackAdversarialConfig, AttackConverterConfig, AttackScoringConfig from pyrit.executor.attack.single_turn.prompt_sending import PromptSendingAttack from pyrit.models import ComponentIdentifier, Identifiable, SeedAttackTechniqueGroup, SeedPrompt from pyrit.prompt_converter import Base64Converter, ROT13Converter diff --git a/tests/unit/scenario/core/test_baseline_deprecation.py b/tests/unit/scenario/core/test_baseline_deprecation.py deleted file mode 100644 index 1e4acde41f..0000000000 --- a/tests/unit/scenario/core/test_baseline_deprecation.py +++ /dev/null @@ -1,215 +0,0 @@ -# Copyright (c) Microsoft Corporation. -# Licensed under the MIT license. - -""" -Deprecated. Will be removed in 0.16.0 along with the corresponding -``include_default_baseline`` / ``include_baseline`` constructor shims in -``Scenario`` and its subclasses (``Cyber``, ``Jailbreak``, ``Scam``, -``RedTeamAgent``, ``Encoding``). -""" - -import warnings -from typing import ClassVar -from unittest.mock import MagicMock, patch - -import pytest - -from pyrit.models import ComponentIdentifier -from pyrit.scenario import DatasetConfiguration -from pyrit.scenario.core import BaselineAttackPolicy, Scenario, ScenarioStrategy -from pyrit.score import TrueFalseScorer - -_TEST_SCORER_ID = ComponentIdentifier(class_name="MockScorer", class_module="tests.unit.scenarios") - - -class _LegacyStrategy(ScenarioStrategy): - TEST = ("test", {"concrete"}) - ALL = ("all", {"all"}) - - @classmethod - def get_aggregate_tags(cls) -> set[str]: - return {"all"} - - -class _LegacyScenario(Scenario): - """Minimal Scenario stand-in for exercising the deprecated baseline kwargs.""" - - BASELINE_ATTACK_POLICY: ClassVar[BaselineAttackPolicy] = BaselineAttackPolicy.Enabled - - def __init__(self, **kwargs): - kwargs.setdefault("strategy_class", _LegacyStrategy) - kwargs.setdefault("default_strategy", _LegacyStrategy.ALL) - kwargs.setdefault("default_dataset_config", DatasetConfiguration()) - if "objective_scorer" not in kwargs: - mock_scorer = MagicMock(spec=TrueFalseScorer) - mock_scorer.get_identifier.return_value = _TEST_SCORER_ID - mock_scorer.get_scorer_metrics.return_value = None - kwargs["objective_scorer"] = mock_scorer - kwargs.setdefault("version", 1) - super().__init__(**kwargs) - - async def _get_atomic_attacks_async(self): - atomic_attacks = [] - if self._include_baseline: - groups_by_dataset = self._dataset_config.get_seed_attack_groups() - all_seed_groups = [g for groups in groups_by_dataset.values() for g in groups] - atomic_attacks.append(self._build_baseline_atomic_attack(seed_groups=all_seed_groups)) - return atomic_attacks - - -@pytest.fixture -def mock_objective_target(): - target = MagicMock() - target.get_identifier.return_value = ComponentIdentifier(class_name="MockTarget", class_module="test") - return target - - -@pytest.mark.usefixtures("patch_central_database") -class TestScenarioBaseDeprecation: - """Cover the deprecated ``Scenario(include_default_baseline=...)`` base kwarg.""" - - def test_base_kwarg_emits_deprecation_warning(self): - with warnings.catch_warnings(record=True) as caught: - warnings.simplefilter("always") - scenario = _LegacyScenario(include_default_baseline=False) - - deprecations = [w for w in caught if issubclass(w.category, DeprecationWarning)] - assert len(deprecations) == 1 - msg = str(deprecations[0].message) - assert "include_default_baseline" in msg - assert "0.16.0" in msg - assert scenario._legacy_include_baseline is False - - def test_base_kwarg_omitted_emits_no_warning(self): - with warnings.catch_warnings(record=True) as caught: - warnings.simplefilter("always") - scenario = _LegacyScenario() - - assert not any(issubclass(w.category, DeprecationWarning) for w in caught) - assert scenario._legacy_include_baseline is None - - async def test_legacy_value_drives_initialize_when_runtime_kwarg_omitted(self, mock_objective_target): - """Constructor-time False suppresses the baseline that BASELINE_ATTACK_POLICY=Enabled would add.""" - with warnings.catch_warnings(): - warnings.simplefilter("ignore", DeprecationWarning) - scenario = _LegacyScenario(include_default_baseline=False) - - with patch.object(_LegacyScenario, "default_dataset_config", create=True, return_value=DatasetConfiguration()): - await scenario.initialize_async(objective_target=mock_objective_target) - - assert not any(a.atomic_attack_name == "baseline" for a in scenario._atomic_attacks) - - async def test_runtime_kwarg_wins_over_legacy_value(self, mock_objective_target): - """Explicit runtime include_baseline overrides any constructor-time legacy value.""" - with warnings.catch_warnings(): - warnings.simplefilter("ignore", DeprecationWarning) - scenario = _LegacyScenario(include_default_baseline=True) - - with patch.object(_LegacyScenario, "default_dataset_config", create=True, return_value=DatasetConfiguration()): - await scenario.initialize_async(objective_target=mock_objective_target, include_baseline=False) - - assert not any(a.atomic_attack_name == "baseline" for a in scenario._atomic_attacks) - - -class TestSubclassBaselineKwargDeprecation: - """Cover the deprecated ``include_baseline`` constructor kwarg on user-facing subclasses.""" - - @pytest.fixture(autouse=True) - def _populate_registry(self): - """Populate the technique registry so Cyber/RapidResponse-style subclasses can build their strategy enum.""" - from pyrit.prompt_target import PromptTarget - from pyrit.registry import TargetRegistry - from pyrit.registry.components.attack_technique_registry import AttackTechniqueRegistry - from pyrit.scenario.scenarios.airt.cyber import Cyber - from pyrit.setup.initializers.components.scenario_techniques import build_scenario_technique_factories - - AttackTechniqueRegistry.reset_registry_singleton() - TargetRegistry.reset_registry_singleton() - Cyber._cached_strategy_class = None - - adv_target = MagicMock(spec=PromptTarget) - adv_target.capabilities.includes.return_value = True - TargetRegistry.get_registry_singleton().instances.register(adv_target, name="adversarial_chat") - - AttackTechniqueRegistry.get_registry_singleton().register_from_factories(build_scenario_technique_factories()) - yield - AttackTechniqueRegistry.reset_registry_singleton() - TargetRegistry.reset_registry_singleton() - Cyber._cached_strategy_class = None - - @pytest.mark.parametrize( - "import_path, class_name, needs_adversarial_chat", - [ - ("pyrit.scenario.scenarios.airt.cyber", "Cyber", False), - ("pyrit.scenario.scenarios.airt.jailbreak", "Jailbreak", False), - ("pyrit.scenario.scenarios.airt.scam", "Scam", True), - ("pyrit.scenario.scenarios.garak.encoding", "Encoding", False), - ], - ) - def test_subclass_kwarg_emits_deprecation_warning( - self, import_path, class_name, needs_adversarial_chat, patch_central_database - ): - from pyrit.prompt_target import PromptTarget - from pyrit.score import TrueFalseScorer - - module = __import__(import_path, fromlist=[class_name]) - cls = getattr(module, class_name) - - # Spec'd against TrueFalseScorer so AttackScoringConfig validators accept it. - mock_scorer = MagicMock(spec=TrueFalseScorer) - mock_scorer.get_identifier.return_value = _TEST_SCORER_ID - mock_scorer.get_scorer_metrics.return_value = None - - extra_kwargs = {} - if needs_adversarial_chat: - mock_target = MagicMock(spec=PromptTarget) - mock_target.get_identifier.return_value = ComponentIdentifier(class_name="MockTarget", class_module="test") - extra_kwargs["adversarial_chat"] = mock_target - - with warnings.catch_warnings(record=True) as caught: - warnings.simplefilter("always") - scenario = cls(objective_scorer=mock_scorer, include_baseline=False, **extra_kwargs) - - deprecations = [ - w for w in caught if issubclass(w.category, DeprecationWarning) and class_name in str(w.message) - ] - assert len(deprecations) >= 1, f"{class_name} did not emit a DeprecationWarning naming the class" - assert "0.16.0" in str(deprecations[0].message) - assert scenario._legacy_include_baseline is False - - -@pytest.mark.usefixtures("patch_central_database") -class TestLegacyAndRuntimePathsEquivalentUnderMaxDatasetSize: - """ADO 9012: the deprecated constructor path and the new initialize_async path must - produce the same baseline atomic attack under max_dataset_size.""" - - async def test_paths_produce_matching_objective_sets(self, mock_objective_target): - from pyrit.models import SeedGroup, SeedObjective - - seed_groups = [SeedGroup(seeds=[SeedObjective(value=f"obj{i}")]) for i in range(10)] - - # Both paths share the same patched sample, so each scenario's single - # resolution call returns ``stable_sample``. - stable_sample = seed_groups[:3] - - with patch( - "pyrit.scenario.core.dataset_configuration.random.sample", - return_value=stable_sample, - ): - config_legacy = DatasetConfiguration(seed_groups=seed_groups, max_dataset_size=3) - with warnings.catch_warnings(): - warnings.simplefilter("ignore", DeprecationWarning) - legacy = _LegacyScenario(include_default_baseline=True) - await legacy.initialize_async(objective_target=mock_objective_target, dataset_config=config_legacy) - - config_runtime = DatasetConfiguration(seed_groups=seed_groups, max_dataset_size=3) - runtime = _LegacyScenario() - await runtime.initialize_async( - objective_target=mock_objective_target, - dataset_config=config_runtime, - include_baseline=True, - ) - - assert legacy._atomic_attacks[0].atomic_attack_name == "baseline" - assert runtime._atomic_attacks[0].atomic_attack_name == "baseline" - assert set(legacy._atomic_attacks[0].objectives) == set(runtime._atomic_attacks[0].objectives) diff --git a/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py b/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py index 092216fe2c..2f77b60fd2 100644 --- a/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py +++ b/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py @@ -374,6 +374,20 @@ def test_drops_strategies_without_factory(self): resolved = resolve_technique_factories(context=context) assert list(resolved.keys()) == ["alpha"] + def test_extra_factories_merged_and_override_registry(self): + registry_factories = {"alpha": _mock_factory(name="alpha")} + local_alpha = _mock_factory(name="alpha") + local_only = _mock_factory(name="local") + context = _context(strategies=[_strategy("alpha"), _strategy("local")]) + with _patch_registry(registry_factories): + resolved = resolve_technique_factories( + context=context, + extra_factories={"alpha": local_alpha, "local": local_only}, + ) + assert list(resolved.keys()) == ["alpha", "local"] + assert resolved["alpha"] is local_alpha # extra overrides the registry factory of the same name + assert resolved["local"] is local_only # local-only factory is selectable without global registration + @pytest.mark.usefixtures("patch_central_database") class TestBuildMatrixAtomicAttacks: @@ -429,3 +443,17 @@ def test_strategy_converters_forwarded(self): extra = factory.create.call_args.kwargs["extra_request_converters"] assert extra is not None assert len(extra) == 1 + + def test_extra_factories_used_for_selection(self): + context = _context( + strategies=[_strategy("local")], + seed_groups_by_dataset={"ds": [_seed_group(objective="o1")]}, + ) + # The selected technique exists only in extra_factories, not the registry. + with _patch_registry({"other": _mock_factory(name="other")}): + result = build_matrix_atomic_attacks( + context=context, + objective_scorer=MagicMock(spec=TrueFalseScorer), + extra_factories={"local": _mock_factory(name="local")}, + ) + assert [a.atomic_attack_name for a in result] == ["local_ds"] diff --git a/tests/unit/scenario/core/test_scenario.py b/tests/unit/scenario/core/test_scenario.py index 276184c3e7..7cfd17a952 100644 --- a/tests/unit/scenario/core/test_scenario.py +++ b/tests/unit/scenario/core/test_scenario.py @@ -17,18 +17,8 @@ from pyrit.executor.attack.core import AttackExecutorResult from pyrit.memory import CentralMemory from pyrit.models import AttackOutcome, AttackResult, ComponentIdentifier -from pyrit.scenario import ( - DatasetAttackConfiguration, - DatasetConfiguration, - ScenarioIdentifier, - ScenarioResult, -) -from pyrit.scenario.core import ( - AtomicAttack, - BaselineAttackPolicy, - Scenario, - ScenarioStrategy, -) +from pyrit.scenario import DatasetConfiguration, ScenarioIdentifier, ScenarioResult +from pyrit.scenario.core import AtomicAttack, BaselineAttackPolicy, Scenario, ScenarioStrategy from pyrit.score import Scorer from tests.unit.mocks import make_scenario_identifier, make_scenario_result @@ -174,10 +164,36 @@ def get_aggregate_tags(cls) -> set[str]: super().__init__(**kwargs) self._atomic_attacks_to_return = atomic_attacks_to_return or [] - async def _get_atomic_attacks_async(self): + async def _resolve_seed_groups_by_dataset_async(self): + return {} + + async def _build_atomic_attacks_async(self, *, context): return self._atomic_attacks_to_return +def test_scenario_base_class_is_abstract(): + """The base ``Scenario`` declares ``_build_atomic_attacks_async`` abstract and can't be instantiated directly.""" + assert "_build_atomic_attacks_async" in Scenario.__abstractmethods__ + with pytest.raises(TypeError, match="_build_atomic_attacks_async"): + Scenario() # type: ignore[abstract] + + +def test_subclass_without_build_atomic_attacks_async_is_abstract(): + """A subclass that omits ``_build_atomic_attacks_async`` stays abstract and fails at instantiation.""" + + class IncompleteScenario(Scenario): + """Subclass that forgets to implement the required extension point.""" + + assert "_build_atomic_attacks_async" in IncompleteScenario.__abstractmethods__ + with pytest.raises(TypeError, match="_build_atomic_attacks_async"): + IncompleteScenario() # type: ignore[abstract] + + +def test_subclass_implementing_build_atomic_attacks_async_is_concrete(): + """Implementing ``_build_atomic_attacks_async`` clears the abstract marker so the subclass is instantiable.""" + assert not ConcreteScenario.__abstractmethods__ + + @pytest.mark.usefixtures("patch_central_database") class TestScenarioInitialization: """Tests for Scenario class initialization.""" @@ -720,23 +736,10 @@ def get_aggregate_tags(cls) -> set[str]: super().__init__(**kwargs) self._atomic_attacks_to_return = atomic_attacks_to_return or [] - async def _get_atomic_attacks_async(self): - atomic_attacks = list(self._atomic_attacks_to_return) - if self._include_baseline: - groups_by_dataset = self._dataset_config.get_seed_attack_groups() - all_seed_groups = [g for groups in groups_by_dataset.values() for g in groups] - atomic_attacks.insert(0, self._build_baseline_atomic_attack(seed_groups=all_seed_groups)) - return atomic_attacks + async def _resolve_seed_groups_by_dataset_async(self): + return self._dataset_config.get_seed_attack_groups() - -class _LegacyOverrideScenario(ConcreteScenarioWithTrueFalseScorer): - """Override that does NOT emit baseline — exercises the deprecation rescue path. - - Real user scenarios written before the structural fix may follow this pattern; - the rescue path warns and injects baseline so they keep working until 0.16.0. - """ - - async def _get_atomic_attacks_async(self): + async def _build_atomic_attacks_async(self, *, context): return list(self._atomic_attacks_to_return) @@ -949,22 +952,14 @@ async def test_baseline_objectives_match_atomic_attacks_under_max_dataset_size( config = DatasetConfiguration(seed_groups=seed_groups, max_dataset_size=3) class StrategyScenario(ConcreteScenarioWithTrueFalseScorer): - async def _get_atomic_attacks_async(self): - groups_by_dataset = self._dataset_config.get_seed_attack_groups() - all_seed_groups = [g for groups in groups_by_dataset.values() for g in groups] - atomic_attacks = [ + async def _build_atomic_attacks_async(self, *, context): + return [ AtomicAttack( atomic_attack_name="strategy", attack_technique=AttackTechnique(attack=MagicMock()), - seed_groups=all_seed_groups, + seed_groups=list(context.seed_groups), ) ] - if self._include_baseline: - atomic_attacks.insert( - 0, - self._build_baseline_atomic_attack(seed_groups=all_seed_groups), - ) - return atomic_attacks # Two distinct samples wired up. A buggy implementation with a second # resolution call would consume both; the structural fix consumes one. @@ -1028,64 +1023,6 @@ def test_raises_when_scorer_is_none(self, mock_objective_target): scenario._build_baseline_atomic_attack(seed_groups=self._seed_groups()) -@pytest.mark.usefixtures("patch_central_database") -class TestBaselineEmissionDeprecationRescue: - """Deprecation rescue (removed in 0.16.0): overrides that don't emit baseline get a - DeprecationWarning + auto-injected baseline so they keep working during the migration. - """ - - @staticmethod - def _dataset_config(): - from pyrit.models import SeedAttackGroup, SeedObjective - - return DatasetAttackConfiguration( - seed_groups=[SeedAttackGroup(seeds=[SeedObjective(value="x")])], - ) - - async def test_rescue_emits_warning_and_injects_baseline(self, mock_objective_target): - import warnings - - scenario = _LegacyOverrideScenario(name="LegacyOverride", version=1) - - with warnings.catch_warnings(record=True) as caught: - warnings.simplefilter("always") - await scenario.initialize_async( - objective_target=mock_objective_target, - dataset_config=self._dataset_config(), - include_baseline=True, - ) - - deprecations = [ - w - for w in caught - if issubclass(w.category, DeprecationWarning) and "_get_atomic_attacks_async" in str(w.message) - ] - assert len(deprecations) == 1, "rescue should emit exactly one DeprecationWarning naming the method" - assert "0.16.0" in str(deprecations[0].message) - assert scenario._atomic_attacks[0].atomic_attack_name == "baseline" - - async def test_well_behaved_override_does_not_trigger_rescue(self, mock_objective_target): - import warnings - - scenario = ConcreteScenarioWithTrueFalseScorer(name="GoodCitizen", version=1) - - with warnings.catch_warnings(record=True) as caught: - warnings.simplefilter("always") - await scenario.initialize_async( - objective_target=mock_objective_target, - dataset_config=self._dataset_config(), - include_baseline=True, - ) - - rescue_warnings = [ - w - for w in caught - if issubclass(w.category, DeprecationWarning) and "_get_atomic_attacks_async" in str(w.message) - ] - assert not rescue_warnings, "well-behaved override should not trigger the rescue path" - assert scenario._atomic_attacks[0].atomic_attack_name == "baseline" - - @pytest.mark.usefixtures("patch_central_database") class TestValidateStoredScenario: """Tests for Scenario._validate_stored_scenario.""" diff --git a/tests/unit/scenario/core/test_scenario_parameters.py b/tests/unit/scenario/core/test_scenario_parameters.py index cb4bb0b953..13c357f923 100644 --- a/tests/unit/scenario/core/test_scenario_parameters.py +++ b/tests/unit/scenario/core/test_scenario_parameters.py @@ -40,7 +40,10 @@ class _ParamTestScenario(Scenario): def supported_parameters(cls) -> list[Parameter]: return list(params_to_declare) - async def _get_atomic_attacks_async(self): + async def _resolve_seed_groups_by_dataset_async(self): + return {} + + async def _build_atomic_attacks_async(self, *, context): return [] mock_scorer = MagicMock(spec=Scorer) diff --git a/tests/unit/scenario/core/test_scenario_partial_results.py b/tests/unit/scenario/core/test_scenario_partial_results.py index fa49a1c340..6181f6f5dd 100644 --- a/tests/unit/scenario/core/test_scenario_partial_results.py +++ b/tests/unit/scenario/core/test_scenario_partial_results.py @@ -109,7 +109,10 @@ def __init__(self, *, atomic_attacks_to_return=None, objective_scorer=None, **kw super().__init__(strategy_class=strategy_class, objective_scorer=objective_scorer, **kwargs) self._test_atomic_attacks = atomic_attacks_to_return or [] - async def _get_atomic_attacks_async(self): + async def _resolve_seed_groups_by_dataset_async(self): + return {} + + async def _build_atomic_attacks_async(self, *, context): return self._test_atomic_attacks diff --git a/tests/unit/scenario/core/test_scenario_retry.py b/tests/unit/scenario/core/test_scenario_retry.py index 53686c58a9..bea48fc98f 100644 --- a/tests/unit/scenario/core/test_scenario_retry.py +++ b/tests/unit/scenario/core/test_scenario_retry.py @@ -180,7 +180,10 @@ def __init__(self, *, atomic_attacks_to_return=None, objective_scorer=None, **kw super().__init__(strategy_class=strategy_class, objective_scorer=objective_scorer, **kwargs) self._atomic_attacks_to_return = atomic_attacks_to_return or [] - async def _get_atomic_attacks_async(self): + async def _resolve_seed_groups_by_dataset_async(self): + return {} + + async def _build_atomic_attacks_async(self, *, context): return self._atomic_attacks_to_return diff --git a/tests/unit/scenario/foundry/test_red_team_agent.py b/tests/unit/scenario/foundry/test_red_team_agent.py index 2a97be6db3..75ba219872 100644 --- a/tests/unit/scenario/foundry/test_red_team_agent.py +++ b/tests/unit/scenario/foundry/test_red_team_agent.py @@ -225,7 +225,7 @@ def test_init_creates_default_scorer_when_not_provided( mock_get_scorer.assert_called_once() assert scenario._attack_scoring_config.objective_scorer == mock_scorer_instance - # seed_groups are resolved lazily during _get_atomic_attacks_async + # seed_groups are resolved lazily during initialize_async assert scenario._attack_scoring_config.objective_scorer == mock_scorer_instance async def test_init_raises_exception_when_no_datasets_available(self, mock_objective_target, mock_objective_scorer): @@ -233,7 +233,7 @@ async def test_init_raises_exception_when_no_datasets_available(self, mock_objec # Don't mock _resolve_seed_groups, let it try to load from empty memory scenario = RedTeamAgent(attack_scoring_config=AttackScoringConfig(objective_scorer=mock_objective_scorer)) - # Error should occur during initialize_async when _get_atomic_attacks_async resolves seed groups. + # Error should occur during initialize_async when it resolves seed groups. # Neutralize the provider fetch so the empty-memory path raises loudly instead of fetching. with patch( "pyrit.scenario.core.dataset_configuration.DatasetConfiguration._fetch_dataset_async", diff --git a/tests/unit/scenario/garak/test_doctor.py b/tests/unit/scenario/garak/test_doctor.py index 94cd9c213b..7449366ae9 100644 --- a/tests/unit/scenario/garak/test_doctor.py +++ b/tests/unit/scenario/garak/test_doctor.py @@ -156,7 +156,7 @@ async def test_atomic_attacks_one_per_technique( dataset_config=doctor_dataset_config, ) - atomic_attacks = await scenario._get_atomic_attacks_async() + atomic_attacks = scenario._atomic_attacks assert len(atomic_attacks) == 2 names = {a.atomic_attack_name for a in atomic_attacks} assert any(n.startswith("policy_puppetry_leet") for n in names) diff --git a/tests/unit/scenario/garak/test_encoding.py b/tests/unit/scenario/garak/test_encoding.py index 555e4eaf98..c127925518 100644 --- a/tests/unit/scenario/garak/test_encoding.py +++ b/tests/unit/scenario/garak/test_encoding.py @@ -11,11 +11,7 @@ from pyrit.models import ComponentIdentifier, SeedAttackGroup, SeedObjective, SeedPrompt from pyrit.prompt_converter import Base64Converter from pyrit.prompt_target import PromptTarget -from pyrit.scenario import ( - CompoundDatasetAttackConfiguration, - DatasetAttackConfiguration, - DatasetConfiguration, -) +from pyrit.scenario import CompoundDatasetAttackConfiguration, DatasetAttackConfiguration, DatasetConfiguration from pyrit.scenario.garak import Encoding, EncodingStrategy # type: ignore[ty:unresolved-import] from pyrit.scenario.scenarios.garak.encoding import EncodingDatasetConfiguration from pyrit.score import DecodingScorer, TrueFalseScorer @@ -231,7 +227,7 @@ async def test_get_atomic_attacks_async_returns_attacks( ) await scenario.initialize_async(objective_target=mock_objective_target, dataset_config=mock_dataset_config) - atomic_attacks = await scenario._get_atomic_attacks_async() + atomic_attacks = scenario._atomic_attacks # Should return multiple atomic attacks (one for each encoding type) assert len(atomic_attacks) > 0 diff --git a/tests/unit/scenario/scenarios/adaptive/test_text_adaptive.py b/tests/unit/scenario/scenarios/adaptive/test_text_adaptive.py index a1fdbd48d1..e4afc2ef03 100644 --- a/tests/unit/scenario/scenarios/adaptive/test_text_adaptive.py +++ b/tests/unit/scenario/scenarios/adaptive/test_text_adaptive.py @@ -15,13 +15,9 @@ from pyrit.models.identifiers import ComponentIdentifier from pyrit.prompt_target import PromptTarget from pyrit.registry.components.attack_technique_registry import AttackTechniqueRegistry -from pyrit.scenario.core.dataset_configuration import ( - CompoundDatasetAttackConfiguration, -) +from pyrit.scenario.core.dataset_configuration import CompoundDatasetAttackConfiguration from pyrit.scenario.core.scenario import BaselineAttackPolicy -from pyrit.scenario.scenarios.adaptive.dispatcher import ( - AdaptiveTechniqueDispatcher, -) +from pyrit.scenario.scenarios.adaptive.dispatcher import AdaptiveTechniqueDispatcher from pyrit.scenario.scenarios.adaptive.text_adaptive import TextAdaptive from pyrit.score import TrueFalseScorer @@ -181,7 +177,7 @@ async def _build_scenario_and_attacks( objective_target=mock_objective_target, include_baseline=False, ) - return scenario, await scenario._get_atomic_attacks_async() + return scenario, scenario._atomic_attacks async def test_one_atomic_per_objective(self, mock_objective_target, mock_objective_scorer): groups = { @@ -226,14 +222,12 @@ def _spy_init(self, *args, **kwargs): return_value=groups, ): scenario = TextAdaptive(objective_scorer=mock_objective_scorer) - await scenario.initialize_async( - objective_target=mock_objective_target, - include_baseline=False, - ) - # Only spy on the explicit invocation so the initialize_async call - # doesn't double-count dispatchers. + # Spy on the dispatcher construction that initialize_async triggers. with patch.object(AdaptiveTechniqueDispatcher, "__init__", _spy_init): - await scenario._get_atomic_attacks_async() + await scenario.initialize_async( + objective_target=mock_objective_target, + include_baseline=False, + ) # One dispatcher per dataset; all share the same selector identity. assert len(selectors_seen) == 2 @@ -277,14 +271,14 @@ async def test_no_usable_techniques_raises(self, mock_objective_target, mock_obj return_value=groups, ): scenario = TextAdaptive(objective_scorer=mock_objective_scorer) - await scenario.initialize_async( - objective_target=mock_objective_target, - include_baseline=False, - ) - # Force the factory map to be empty. + # Force the factory map to be empty; initialize_async builds the atomic + # attacks and must raise when no techniques are usable. with patch.object(scenario, "_get_attack_technique_factories", return_value={}): with pytest.raises(ValueError, match="no usable techniques"): - await scenario._get_atomic_attacks_async() + await scenario.initialize_async( + objective_target=mock_objective_target, + include_baseline=False, + ) async def test_techniques_with_seed_technique_are_kept(self, mock_objective_target, mock_objective_scorer): """Factories that declare a ``seed_technique`` participate in the pool