diff --git a/.github/instructions/scenarios.instructions.md b/.github/instructions/scenarios.instructions.md index 59a18dd835..9089b3f6ad 100644 --- a/.github/instructions/scenarios.instructions.md +++ b/.github/instructions/scenarios.instructions.md @@ -237,7 +237,7 @@ async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list `build_matrix_atomic_attacks`: 1. Calls `resolve_technique_factories(context=context)` to map the selected techniques to their registered `AttackTechniqueFactory` instances (reads the `AttackTechniqueRegistry` singleton; - techniques with no registered factory are dropped). + raises ``TechniqueResolutionError`` if any selected technique has no registered factory). 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 `--techniques :converter.`) as diff --git a/pyrit/scenario/core/matrix_atomic_attack_builder.py b/pyrit/scenario/core/matrix_atomic_attack_builder.py index 148b098be3..34639660ec 100644 --- a/pyrit/scenario/core/matrix_atomic_attack_builder.py +++ b/pyrit/scenario/core/matrix_atomic_attack_builder.py @@ -29,6 +29,16 @@ from pyrit.scenario.core.atomic_attack import AtomicAttack from pyrit.scenario.core.attack_technique import AttackTechnique + +class TechniqueResolutionError(ValueError): + """ + Raised when a selected scenario technique has no registered factory. + + Subclasses ``ValueError`` so existing ``except ValueError`` handlers keep working, + mirroring ``DatasetConstraintError``. + """ + + if TYPE_CHECKING: from collections.abc import Callable, Mapping, Sequence @@ -142,8 +152,8 @@ def resolve_technique_factories( Resolve a run's selected techniques to their registered ``AttackTechniqueFactory`` instances. Reads the ``AttackTechniqueRegistry`` singleton and keeps only the factories whose name - matches a selected technique, preserving selection order. Techniques with no registered - factory are silently dropped so the caller can proceed with whatever techniques exist. + matches a selected technique, preserving selection order. Raises if any selected + technique has no registered factory so the run cannot silently omit requested work. Args: context (ScenarioContext): The resolved runtime inputs for this run. @@ -155,6 +165,9 @@ def resolve_technique_factories( Returns: dict[str, AttackTechniqueFactory]: Mapping of technique name to factory, ordered by the selected techniques. + + Raises: + TechniqueResolutionError: If any selected technique has no registered factory. """ return resolve_technique_factories_for_techniques( scenario_techniques=context.scenario_techniques, @@ -170,19 +183,33 @@ def resolve_technique_factories_for_techniques( """ Resolve selected concrete techniques to their canonical factories. + Args: + scenario_techniques (Sequence[ScenarioTechnique]): Concrete techniques to resolve. + extra_factories (dict[str, AttackTechniqueFactory] | None): Scenario-local factories + merged on top of the registry. + Returns: dict[str, AttackTechniqueFactory]: Selected factories in technique order. + + Raises: + TechniqueResolutionError: If any selected technique has no registered factory. """ from pyrit.registry.components.attack_technique_registry import AttackTechniqueRegistry all_factories = dict(AttackTechniqueRegistry.get_registry_singleton().get_factories_or_raise()) if extra_factories: all_factories.update(extra_factories) - return { - technique.value: all_factories[technique.value] - for technique in scenario_techniques - if technique.value in all_factories - } + + missing = list(dict.fromkeys(t.value for t in scenario_techniques if t.value not in all_factories)) + + if missing: + raise TechniqueResolutionError( + "The following selected attack techniques have no registered factory: " + f"{', '.join(missing)}. Register the techniques (or pass them via " + "extra_factories) before starting the run." + ) + + return {technique.value: all_factories[technique.value] for technique in scenario_techniques} def filter_compatible_seed_groups( diff --git a/pyrit/scenario/scenarios/airt/jailbreak.py b/pyrit/scenario/scenarios/airt/jailbreak.py index cad7fe6f17..9f60c53f39 100644 --- a/pyrit/scenario/scenarios/airt/jailbreak.py +++ b/pyrit/scenario/scenarios/airt/jailbreak.py @@ -430,13 +430,6 @@ async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list num_attempts = self.params.get("num_jailbreak_attempts", 1) technique_factories = resolve_technique_factories(context=context, extra_factories=_extra_default_factories()) - selected_names = {technique.value for technique in context.scenario_techniques} - missing = selected_names - set(technique_factories) - if missing: - raise ValueError( - "Jailbreak selected techniques that are no longer available: " - f"{sorted(missing)}. Refresh the plan and select a supported delivery method." - ) prompt_sending_factory = technique_factories.get(_PROMPT_SENDING) system_selected = _JAILBREAK_SYSTEM_PROMPT in technique_factories diff --git a/tests/unit/scenario/airt/test_jailbreak.py b/tests/unit/scenario/airt/test_jailbreak.py index cf0f25c436..e94438d086 100644 --- a/tests/unit/scenario/airt/test_jailbreak.py +++ b/tests/unit/scenario/airt/test_jailbreak.py @@ -442,16 +442,6 @@ async def test_stale_incompatible_technique_is_rejected( with pytest.raises(ValueError, match="stale or incompatible"): await scenario.initialize_async() - async def test_missing_runtime_factory_is_rejected( - self, mock_objective_target, mock_objective_scorer, mock_memory_seed_groups - ): - with _patch_seed_groups(mock_memory_seed_groups): - with patch("pyrit.scenario.scenarios.airt.jailbreak.resolve_technique_factories", return_value={}): - scenario = Jailbreak(objective_scorer=mock_objective_scorer) - scenario.set_params_from_args(args=_default_args(mock_objective_target, jailbreak_names=["aim.yaml"])) - with pytest.raises(ValueError, match="no longer available.*prompt_sending"): - await scenario.initialize_async() - async def test_all_templates_produce_attacks( self, mock_objective_target, mock_objective_scorer, mock_memory_seed_groups ): 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 044ce78512..f5d46b65af 100644 --- a/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py +++ b/tests/unit/scenario/core/test_matrix_atomic_attack_builder.py @@ -26,6 +26,7 @@ from pyrit.scenario.core.matrix_atomic_attack_builder import ( MatrixAtomicAttackBuilder, MatrixCombo, + TechniqueResolutionError, build_baseline_atomic_attack, build_matrix_atomic_attacks, resolve_technique_factories, @@ -401,12 +402,39 @@ def test_keeps_only_selected_in_order(self): resolved = resolve_technique_factories(context=context) assert list(resolved.keys()) == ["beta", "alpha"] - def test_drops_techniques_without_factory(self): + def test_raises_when_any_selected_technique_is_missing(self): factories = {"alpha": _mock_factory(name="alpha")} context = _context(techniques=[_technique("alpha"), _technique("missing")]) - with _patch_registry(factories): - resolved = resolve_technique_factories(context=context) - assert list(resolved.keys()) == ["alpha"] + with _patch_registry(factories), pytest.raises(TechniqueResolutionError, match="missing"): + resolve_technique_factories(context=context) + + def test_raises_when_all_selected_techniques_missing(self): + """A nonempty selection resolving to nothing must fail loudly, not run baseline-only.""" + factories = {"alpha": _mock_factory(name="alpha")} + context = _context(techniques=[_technique("missing_a"), _technique("missing_b")]) + with _patch_registry(factories), pytest.raises(TechniqueResolutionError, match="missing_a"): + resolve_technique_factories(context=context) + + def test_empty_selection_resolves_without_error(self): + context = _context(techniques=[]) + with _patch_registry({}): + assert resolve_technique_factories(context=context) == {} + + def test_error_lists_each_missing_technique_once_in_selection_order(self): + factories = {"alpha": _mock_factory(name="alpha")} + context = _context( + techniques=[ + _technique("missing_a"), + _technique("alpha"), + _technique("missing_b"), + _technique("missing_a"), + ] + ) + with _patch_registry(factories), pytest.raises(TechniqueResolutionError) as exc_info: + resolve_technique_factories(context=context) + message = str(exc_info.value) + assert message.index("missing_a") < message.index("missing_b") + assert message.count("missing_a") == 1 def test_extra_factories_merged_and_override_registry(self): registry_factories = {"alpha": _mock_factory(name="alpha")}