diff --git a/.github/instructions/scenarios.instructions.md b/.github/instructions/scenarios.instructions.md index 640ec9b544..044275ab4d 100644 --- a/.github/instructions/scenarios.instructions.md +++ b/.github/instructions/scenarios.instructions.md @@ -16,7 +16,7 @@ All scenarios inherit from `Scenario` (ABC) and must: 2. **Optionally declare `BASELINE_ATTACK_POLICY`** (defaults to `BaselineAttackPolicy.Enabled` — a baseline `PromptSendingAttack` is prepended and callers can opt out per run by setting `include_baseline=False` in the run params, see "Run Parameters" below): - `BaselineAttackPolicy.Disabled` — baseline supported but off by default (e.g. `Jailbreak`, where templates dominate the run). - `BaselineAttackPolicy.Forbidden` — baseline is meaningless for this scenario's comparison axis (e.g. `AdversarialBenchmark`, which compares against gold-standard answers). Supplying `include_baseline=True` raises `ValueError`. -3. **Pass `technique_class`, `default_technique`, and `default_dataset_config` to `super().__init__()`:** +3. **Pass `technique_class` and `default_dataset_config` to `super().__init__()`:** ```python class MyScenario(Scenario): @@ -28,13 +28,16 @@ class MyScenario(Scenario): super().__init__( version=self.VERSION, technique_class=MyTechnique, - default_technique=MyTechnique.ALL, default_dataset_config=DatasetConfiguration(dataset_names=["my_dataset"]), objective_scorer=objective_scorer or self._get_default_objective_scorer(), scenario_result_id=scenario_result_id, ) ``` +The default technique (what runs when the caller passes no `scenario_techniques`) is +**not** a constructor argument — it is owned by the technique catalog and read from +`technique_class.default()`. See "Default Technique" below. + For scenarios whose technique enum is built dynamically (RapidResponse pattern), build the technique class in a module-level `@cache`-decorated function and pass the result through the constructor — no classmethod indirection required. @@ -65,7 +68,6 @@ def __init__( super().__init__( version=self.VERSION, technique_class=MyTechnique, - default_technique=MyTechnique.ALL, default_dataset_config=DatasetConfiguration(dataset_names=["my_dataset"]), objective_scorer=objective_scorer, ) @@ -78,7 +80,7 @@ Requirements: `pyrit/common/brick_contract.py`). Violators raise `TypeError` at import time. - **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`, `technique_class`, `default_technique`, `default_dataset_config`, `objective_scorer` +- `super().__init__()` called with `version`, `technique_class`, `default_dataset_config`, `objective_scorer` (the default technique is owned by the catalog, not passed here — see "Default Technique") - complex objects like `adversarial_chat` or `objective_scorer` should be passed into the constructor. ## Run Parameters @@ -153,12 +155,34 @@ class MyTechnique(ScenarioTechnique): @classmethod def get_aggregate_tags(cls) -> set[str]: return {"all", "default", "single_turn", "multi_turn"} + + @classmethod + def default(cls) -> "MyTechnique": + return cls.DEFAULT # what runs when the caller selects nothing ``` - `ALL` aggregate is always required - Each member: `NAME = ("string_value", {tag_set})` - Aggregates expand to all techniques matching their tag +### Default Technique + +The default technique — what runs when no `scenario_techniques` are passed — is a property +of the **technique catalog**, not the scenario. It is *not* passed to `super().__init__()`; +the base `Scenario` reads it from `technique_class.default()`. + +- **Statically declared enums:** override the `default()` classmethod to return the default + member (e.g. `return cls.DEFAULT`, `return cls.EASY`). If the default is `ALL`, don't + override — the base `ScenarioTechnique.default()` already falls back to `ALL`. +- **Dynamically built enums** (`build_technique_class_from_factories`, the RapidResponse + pattern): declare the default via exactly one of `default_tags={...}` (every pool technique + carrying any of those tags) or `default_names={...}` (exact technique names). Both build the + synthetic `DEFAULT` aggregate and record it so `default()` returns it. Passing both raises + `ValueError`. When neither is given (or nothing matches), the default falls back to `ALL`. + +The default is scenario-relative: the same catalog technique can be the default for one +scenario and not another, so there is deliberately no catalog-wide `default` tag. + ### Result grouping (`display_group`) `display_group` controls how attack results are aggregated for display. It is set per @@ -248,9 +272,11 @@ AttackTechniqueFactory( Key points: - `name` is required and must match the technique enum value the scenario looks up. -- `technique_tags` on the factory drives `TagQuery` filters used by - `AttackTechniqueRegistry.build_technique_class_from_factories(...)`. This is **distinct** - from the per-entry `tags` argument passed to `registry.register_technique(...)`. +- `technique_tags` on the factory drives the aggregates and default selection built by + `AttackTechniqueRegistry.build_technique_class_from_factories(...)`: every tag present in + the pool auto-promotes to a selectable aggregate, and `default_tags` picks the default set + by tag. This is **distinct** from the per-entry `tags` argument passed to + `registry.register_technique(...)`. - `uses_adversarial` is auto-derived from the attack class signature (presence of `attack_adversarial_config`) and seed shape; pass `False` explicitly to opt out, or `True` to force opt-in. diff --git a/doc/code/framework.md b/doc/code/framework.md index aa6d0d0c94..4b53b5fcb8 100644 --- a/doc/code/framework.md +++ b/doc/code/framework.md @@ -177,7 +177,7 @@ If you are contributing to PyRIT, that work will most likely land in one of the - Techniques are self-describing `AttackTechniqueFactory` instances (a `name`, `attack_class`, `attack_kwargs`, and `technique_tags`). They read like configuration even though they are code, so adding one — or bringing your own — is easy. The canonical catalog lives under `pyrit/setup/initializers/techniques/`: `core.py` (a small, curated standard set, auto-loaded on a bare `initialize_pyrit_async`), `extra.py` (the broader collection, opt-in), and per-source modules like `airt.py` (owned by a source/scenario but reusable, tagged with their owner and kept out of the default pool). - `core` stays deliberately small so a default run doesn't print 200 techniques or take forever; the wider catalog lives in `extra` and is selected on demand. Users pick subsets by passing initializer tags (e.g. `core`, `extra`, `all`) or writing their own initializer, so different runs — including from the CLI — can register different technique sets without changing the catalog. - A technique tied to one scenario is fine; if it's pinned and non-reusable it can stay local to that scenario, but if another scenario could reuse it, promote it to a catalog module and tag it. -- Tags describe a technique (behavioral tags like `single_turn`/`multi_turn`, owner tags like `airt`); they don't decide what a scenario runs. There is deliberately **no global `default` tag** — a default is scenario-relative, declared per scenario via `build_technique_class_from_factories` (`available` selects the pool, aggregates are named presets, `default` / `default_technique_names` set what runs when nothing is chosen). +- Tags describe a technique (behavioral tags like `single_turn`/`multi_turn`, owner tags like `airt`); they don't decide what a scenario runs. There is deliberately **no global `default` tag** — a default is scenario-relative, declared per scenario via `build_technique_class_from_factories` (the `factories` list is the pool, catalog tags become named aggregate presets, and `default_tags` / `default_names` set what runs when nothing is chosen). - **Does not own**: the conversation algorithm itself. Branching, turn management, and scoring decisions live in the executor it wraps — a technique only selects and configures existing components, and shouldn't implement new sending, scoring, or branching logic. **Framework Plans**: diff --git a/doc/code/scenarios/0_scenarios.ipynb b/doc/code/scenarios/0_scenarios.ipynb index 82d2c1bc76..a20a1e9373 100644 --- a/doc/code/scenarios/0_scenarios.ipynb +++ b/doc/code/scenarios/0_scenarios.ipynb @@ -57,10 +57,10 @@ " - 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 technique that expands to all available techniques\n", + " - The default technique (what runs when the caller selects nothing) is owned by the catalog, not the scenario: override the `default()` classmethod to return the default member (omit it to fall back to `ALL`)\n", "\n", "2. **Scenario Class**: Extend `Scenario` and pass these to `super().__init__()`:\n", " - `technique_class`: Your technique enum class\n", - " - `default_technique`: The default technique (typically `YourTechnique.ALL` or `YourTechnique.DEFAULT`)\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", @@ -72,7 +72,6 @@ " - `name`: Descriptive name for your scenario\n", " - `version`: Integer version number\n", " - `technique_class`: The technique enum class for this scenario\n", - " - `default_technique`: The default technique member (typically `YourTechnique.ALL` or `YourTechnique.DEFAULT`)\n", " - `default_dataset_config`: A `DatasetConfiguration` specifying the scenario's default datasets\n", " - `objective_scorer`: The scorer used to judge responses\n", " - `scenario_result_id`: Optional ID to resume an existing scenario (optional)\n", @@ -141,6 +140,10 @@ " PromptSending = (\"prompt_sending\", {\"single_turn\", \"default\"})\n", " RolePlay = (\"role_play_movie_script\", {\"single_turn\"})\n", "\n", + " @classmethod\n", + " def default(cls) -> \"MyTechnique\":\n", + " return cls.DEFAULT\n", + "\n", "\n", "class MyScenario(Scenario):\n", " \"\"\"Quick-check scenario for testing model behavior across harm categories.\"\"\"\n", @@ -162,7 +165,6 @@ " version=self.VERSION,\n", " objective_scorer=self._objective_scorer,\n", " technique_class=MyTechnique,\n", - " default_technique=MyTechnique.DEFAULT,\n", " default_dataset_config=DatasetConfiguration(dataset_names=[\"dataset_name\"], max_dataset_size=4),\n", " scenario_result_id=scenario_result_id,\n", " )\n", diff --git a/doc/code/scenarios/0_scenarios.py b/doc/code/scenarios/0_scenarios.py index 1a6fc2a0bf..4632ea5612 100644 --- a/doc/code/scenarios/0_scenarios.py +++ b/doc/code/scenarios/0_scenarios.py @@ -59,10 +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 technique that expands to all available techniques +# - The default technique (what runs when the caller selects nothing) is owned by the +# catalog, not the scenario: override the `default()` classmethod to return the default +# member (omit it to fall back to `ALL`) # # 2. **Scenario Class**: Extend `Scenario` and pass these to `super().__init__()`: # - `technique_class`: Your technique enum class -# - `default_technique`: The default technique (typically `YourTechnique.ALL` or `YourTechnique.DEFAULT`) # - Implement `_build_atomic_attacks_async(context)` — the single abstract extension point. # Matrix-shaped scenarios delegate to `build_matrix_atomic_attacks(context=...)` in one line. # @@ -74,7 +76,6 @@ # - `name`: Descriptive name for your scenario # - `version`: Integer version number # - `technique_class`: The technique enum class for this scenario -# - `default_technique`: The default technique member (typically `YourTechnique.ALL` or `YourTechnique.DEFAULT`) # - `default_dataset_config`: A `DatasetConfiguration` specifying the scenario's default datasets # - `objective_scorer`: The scorer used to judge responses # - `scenario_result_id`: Optional ID to resume an existing scenario (optional) @@ -118,6 +119,10 @@ class MyTechnique(ScenarioTechnique): PromptSending = ("prompt_sending", {"single_turn", "default"}) RolePlay = ("role_play_movie_script", {"single_turn"}) + @classmethod + def default(cls) -> "MyTechnique": + return cls.DEFAULT + class MyScenario(Scenario): """Quick-check scenario for testing model behavior across harm categories.""" @@ -139,7 +144,6 @@ def __init__( version=self.VERSION, objective_scorer=self._objective_scorer, technique_class=MyTechnique, - default_technique=MyTechnique.DEFAULT, default_dataset_config=DatasetConfiguration(dataset_names=["dataset_name"], max_dataset_size=4), scenario_result_id=scenario_result_id, ) diff --git a/doc/scanner/1_pyrit_scan.ipynb b/doc/scanner/1_pyrit_scan.ipynb index 42e745b5f4..06ab744d33 100644 --- a/doc/scanner/1_pyrit_scan.ipynb +++ b/doc/scanner/1_pyrit_scan.ipynb @@ -941,7 +941,6 @@ " version=1,\n", " objective_scorer=TrueFalseInverterScorer(scorer=SelfAskRefusalScorer(chat_target=OpenAIChatTarget())),\n", " technique_class=MyCustomTechnique,\n", - " default_technique=MyCustomTechnique.ALL,\n", " default_dataset_config=DatasetAttackConfiguration(dataset_names=[\"harmbench\"]),\n", " scenario_result_id=scenario_result_id,\n", " )\n", diff --git a/doc/scanner/1_pyrit_scan.py b/doc/scanner/1_pyrit_scan.py index 522d1d8dd2..60d2b3937d 100644 --- a/doc/scanner/1_pyrit_scan.py +++ b/doc/scanner/1_pyrit_scan.py @@ -185,7 +185,6 @@ def __init__(self, *, scenario_result_id=None, **kwargs): version=1, objective_scorer=TrueFalseInverterScorer(scorer=SelfAskRefusalScorer(chat_target=OpenAIChatTarget())), technique_class=MyCustomTechnique, - default_technique=MyCustomTechnique.ALL, default_dataset_config=DatasetAttackConfiguration(dataset_names=["harmbench"]), scenario_result_id=scenario_result_id, ) diff --git a/pyrit/registry/components/attack_technique_registry.py b/pyrit/registry/components/attack_technique_registry.py index ffa0bbbca0..c185f5e33f 100644 --- a/pyrit/registry/components/attack_technique_registry.py +++ b/pyrit/registry/components/attack_technique_registry.py @@ -29,7 +29,6 @@ from pyrit.registry.registry_metadata import RegistryMetadata if TYPE_CHECKING: - from pyrit.registry.tag_query import TagQuery from pyrit.scenario.core.attack_technique_factory import ( AttackTechniqueFactory, ScorerOverridePolicy, @@ -177,93 +176,95 @@ def build_technique_class_from_factories( *, class_name: str, factories: list[AttackTechniqueFactory], - aggregate_tags: dict[str, TagQuery], - available: TagQuery | None = None, - default: TagQuery | None = None, - default_technique_names: set[str] | None = None, + default_tags: set[str] | None = None, + default_names: set[str] | None = None, ) -> type: """ Build a ``ScenarioTechnique`` enum subclass dynamically from technique factories. Creates an enum class with: - An ``ALL`` aggregate member (always included). - - A ``DEFAULT`` aggregate member when a default selection is provided. - - Additional aggregate members from ``aggregate_tags`` keys. - - One technique member per *available* factory, with tags from the factory. - - The three selection roles are all expressed the same way — as tag queries - over ``factories`` — and relate as strict subsets: - - - **available** (the pool): ``available`` filters ``factories`` to the - techniques this scenario exposes. When ``None`` the whole ``factories`` - list is the pool (back-compatible). - - **aggregates**: named ``TagQuery`` presets (e.g. ``single_turn``); each is - evaluated only over the pool, so every aggregate is a subset of available. - - **default**: what runs when the caller selects nothing. Given as a - ``TagQuery`` (``default``) and/or explicit ``default_technique_names``; - both are intersected with the pool, so ``DEFAULT`` is always a subset of - available. - - ``default`` is deliberately **not** an intrinsic technique tag: what runs by - default differs per scenario. A scenario selects its default set via a query - or by name so the same technique can be default for one scenario and not - another, without a catalog-wide tag. + - A ``DEFAULT`` aggregate member when a default selection is provided and at + least one pool technique matches it. + - An aggregate member for every catalog tag present in the pool, so tags and + aggregates are synonymous: selecting a tag (e.g. ``core`` or a custom + ``airt_internal``) expands to every technique carrying it. + - One technique member per factory, with tags from the factory. + + The catalog's *default* — what runs when the caller selects nothing — is defined + by exactly one of ``default_tags`` or ``default_names`` (or neither). Both build + the synthetic ``DEFAULT`` aggregate and are recorded on the class, returned by + ``ScenarioTechnique.default()``. When neither is given, or the chosen set matches + no pool technique (e.g. a custom initializer registers no ``light``-tagged + factory), the default falls back to ``ALL``. The default is chosen per-scenario, + so the same technique can be the default for one scenario and not another. Args: class_name (str): Name for the generated enum class. - factories (list[AttackTechniqueFactory]): Candidate technique factories. - Filtered by ``available`` to form the pool of enum members. - aggregate_tags (dict[str, TagQuery]): Maps aggregate member names to a - ``TagQuery`` that selects which pool techniques belong to the aggregate. - An ``ALL`` aggregate (expanding to all pool techniques) is always added. - available (TagQuery | None): Query selecting which of ``factories`` are - available for this scenario (the pool). ``None`` means all of them. - default (TagQuery | None): Query selecting the pool techniques that form - the ``DEFAULT`` aggregate. Combined (union) with - ``default_technique_names``. - default_technique_names (set[str] | None): Names of pool techniques that - form this scenario's ``DEFAULT`` aggregate. Combined (union) with - ``default``. Names not present in the pool are ignored, so a scenario - can list its intended default set even when some of those techniques - are filtered out of its pool. When the combined default selection is - empty, no ``DEFAULT`` aggregate is generated. + factories (list[AttackTechniqueFactory]): The technique factories that form + this scenario's pool of enum members. Callers pre-filter this list to + shape the pool. + default_tags (set[str] | None): Tags whose union defines the scenario's + ``DEFAULT`` aggregate — every pool technique carrying any of these tags is + the default (e.g. ``{"light"}``). Mutually exclusive with ``default_names``. + default_names (set[str] | None): Exact technique names that form the + scenario's ``DEFAULT`` aggregate. Names not present in the pool are + ignored, so a scenario can list its intended default set even when some of + those techniques are filtered out. Mutually exclusive with ``default_tags``. Returns: type: A ``ScenarioTechnique`` subclass with the generated members. + + Raises: + ValueError: If both ``default_tags`` and ``default_names`` are provided. """ from pyrit.scenario import ScenarioTechnique - # available (the pool): filter the candidate factories by the availability query. - pool = available.filter(factories) if available is not None else list(factories) + if default_tags and default_names: + raise ValueError("Provide at most one of default_tags or default_names, not both.") - # default: resolve from an explicit name set and/or a query over the pool. The - # DEFAULT aggregate exists whenever a default was requested; its membership is - # limited to the pool below (only pool factories are iterated), so DEFAULT is - # always a subset of available. - default_names: set[str] = set(default_technique_names or set()) - if default is not None: - default_names |= {f.name for f in pool if default.matches(set(f.technique_tags))} + pool = list(factories) + pool_technique_names = {f.name for f in pool} + pool_tags = {tag for f in pool for tag in f.technique_tags} - all_aggregate_tag_names = {"all"} | set(aggregate_tags.keys()) + # default: the pool techniques that form the DEFAULT aggregate, from either an + # explicit set of names or the union over a set of tags. Limited to the pool, so + # DEFAULT is always a subset of ALL. When it is empty (nothing matched) no DEFAULT + # aggregate is built and the catalog default falls back to ALL. if default_names: + default_member_names = {f.name for f in pool if f.name in default_names} + elif default_tags: + default_member_names = {f.name for f in pool if set(f.technique_tags) & default_tags} + else: + default_member_names = set() + + # Auto-promote every catalog tag present in the pool into a selectable aggregate, + # so tags and aggregates are synonymous: selecting a tag expands to every technique + # carrying it. "all" and "default" are reserved synthetic aggregates and are never + # derived from tags. A tag that collides with a technique name stays a concrete + # technique (name selection wins). + reserved_aggregate_tags = {"all", "default"} + auto_aggregate_tags = pool_tags - reserved_aggregate_tags - pool_technique_names + + all_aggregate_tag_names = {"all"} | auto_aggregate_tags + if default_member_names: all_aggregate_tag_names.add("default") members: dict[str, tuple[str, set[str]]] = {} # Aggregate members first (ALL is always present) members["ALL"] = ("all", {"all"}) - if default_names: + if default_member_names: members["DEFAULT"] = ("default", {"default"}) - for agg_name in aggregate_tags: + for agg_name in sorted(auto_aggregate_tags): members[agg_name.upper()] = (agg_name, {agg_name}) - # Technique members from the pool — assign aggregate tags based on TagQuery matching + # Technique members from the pool — tag DEFAULT members so the aggregate expands. for factory in pool: factory_tags = set(factory.technique_tags) - matched_agg_tags = {agg_name for agg_name, query in aggregate_tags.items() if query.matches(factory_tags)} - if factory.name in default_names: - matched_agg_tags.add("default") - members[factory.name] = (factory.name, factory_tags | matched_agg_tags) + if factory.name in default_member_names: + factory_tags = factory_tags | {"default"} + members[factory.name] = (factory.name, factory_tags) # Build the enum class dynamically technique_cls = ScenarioTechnique(class_name, members) @@ -275,6 +276,12 @@ def _get_aggregate_tags(cls: type) -> set[str]: technique_cls.get_aggregate_tags = _get_aggregate_tags # type: ignore[ty:invalid-assignment] + # Record the catalog's default only when a DEFAULT aggregate was actually built. + # When it wasn't, the attribute is left unset and ScenarioTechnique.default() owns + # the single ALL fallback — so the "no default -> ALL" rule lives in one place. + if default_member_names: + technique_cls._default_technique_value = "default" # type: ignore[ty:unresolved-attribute] + return technique_cls # type: ignore[ty:invalid-return-type] def register_from_factories( diff --git a/pyrit/scenario/core/scenario.py b/pyrit/scenario/core/scenario.py index 7f5a9927c9..bbee5b09d3 100644 --- a/pyrit/scenario/core/scenario.py +++ b/pyrit/scenario/core/scenario.py @@ -153,7 +153,6 @@ def __init__( name: str = "", version: int, technique_class: type[ScenarioTechnique], - default_technique: ScenarioTechnique, default_dataset_config: DatasetAttackConfiguration, objective_scorer: Scorer, scenario_result_id: uuid.UUID | str | None = None, @@ -165,9 +164,9 @@ def __init__( name (str): Descriptive name for the scenario. version (int): Version number of the scenario. technique_class (type[ScenarioTechnique]): The technique enum class for this scenario. - default_technique (ScenarioTechnique): The default technique member used when no - ``scenario_techniques`` are passed to ``initialize_async``. Usually an aggregate - member like ``MyTechnique.ALL`` or ``MyTechnique.DEFAULT``. + The technique used when no ``scenario_techniques`` are passed to + ``initialize_async`` is the catalog's own default, read from + ``technique_class.default()``. default_dataset_config (DatasetAttackConfiguration): The default dataset configuration used when no ``dataset_config`` is passed to ``initialize_async``. objective_scorer (Scorer): The objective scorer used to evaluate attack results. @@ -197,7 +196,7 @@ def __init__( # Store technique configuration for use in initialize_async self._technique_class = technique_class - self._default_technique = default_technique + self._default_technique = technique_class.default() self._default_dataset_config = default_dataset_config # These will be set in initialize_async diff --git a/pyrit/scenario/core/scenario_technique.py b/pyrit/scenario/core/scenario_technique.py index 07550cc22f..dfcf1a8486 100644 --- a/pyrit/scenario/core/scenario_technique.py +++ b/pyrit/scenario/core/scenario_technique.py @@ -114,6 +114,26 @@ def tags(self) -> set[str]: """ return self._tags + @classmethod + def default(cls: type[T]) -> T: + """ + Return the technique member used when the caller selects nothing. + + The default is a property of the technique catalog, not of the scenario that + consumes it. Dynamically built classes have their default recorded by + ``build_technique_class_from_factories`` (stored as ``_default_technique_value``). + Statically declared subclasses whose default is not ``ALL`` override this + classmethod to return their default member; everything else falls back to ``ALL``, + which every catalog always contains. + + Returns: + T: The default technique member. + """ + value = getattr(cls, "_default_technique_value", None) + if value is not None: + return cls(value) + return cls["ALL"] + @classmethod def get_aggregate_tags(cls: type[T]) -> set[str]: """ diff --git a/pyrit/scenario/scenarios/adaptive/adaptive_scenario.py b/pyrit/scenario/scenarios/adaptive/adaptive_scenario.py index 5a9b812950..6916a7189e 100644 --- a/pyrit/scenario/scenarios/adaptive/adaptive_scenario.py +++ b/pyrit/scenario/scenarios/adaptive/adaptive_scenario.py @@ -71,12 +71,6 @@ def get_technique_class(cls) -> type[ScenarioTechnique]: """Return the scenario's technique enum (subclasses must override).""" raise NotImplementedError - @classmethod - @abstractmethod - def get_default_technique(cls) -> ScenarioTechnique: - """Return the scenario's default technique aggregate (subclasses must override).""" - raise NotImplementedError - @classmethod @abstractmethod def default_dataset_config(cls) -> DatasetAttackConfiguration: @@ -108,7 +102,6 @@ def __init__( super().__init__( version=self.VERSION, technique_class=self.get_technique_class(), - default_technique=self.get_default_technique(), default_dataset_config=self.default_dataset_config(), objective_scorer=objective_scorer, scenario_result_id=scenario_result_id, diff --git a/pyrit/scenario/scenarios/adaptive/text_adaptive.py b/pyrit/scenario/scenarios/adaptive/text_adaptive.py index 30d616516b..ff31ed29d8 100644 --- a/pyrit/scenario/scenarios/adaptive/text_adaptive.py +++ b/pyrit/scenario/scenarios/adaptive/text_adaptive.py @@ -19,7 +19,6 @@ from pyrit.common import apply_defaults from pyrit.models.parameter import Parameter 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.scenarios.adaptive.adaptive_scenario import AdaptiveScenario @@ -71,11 +70,7 @@ def _build_text_adaptive_technique() -> type[ScenarioTechnique]: return AttackTechniqueRegistry.build_technique_class_from_factories( # type: ignore[return-value, ty:invalid-return-type] class_name="TextAdaptiveTechnique", factories=factories, - aggregate_tags={ - "single_turn": TagQuery.any_of("single_turn"), - "multi_turn": TagQuery.any_of("multi_turn"), - }, - default_technique_names={"role_play_movie_script", "many_shot"}, + default_names={"role_play_movie_script", "many_shot"}, ) @@ -104,12 +99,6 @@ def get_technique_class(cls) -> type[ScenarioTechnique]: cls._cached_technique_class = _build_text_adaptive_technique() return cls._cached_technique_class - @classmethod - def get_default_technique(cls) -> ScenarioTechnique: - """Return the default technique aggregate (resolves to every ``default``-tagged technique).""" - technique_class = cls.get_technique_class() - return technique_class("default") - @classmethod def required_datasets(cls) -> list[str]: """Return the dataset names this scenario expects when no override is provided.""" diff --git a/pyrit/scenario/scenarios/airt/cyber.py b/pyrit/scenario/scenarios/airt/cyber.py index 51debd1df9..22022bba0c 100644 --- a/pyrit/scenario/scenarios/airt/cyber.py +++ b/pyrit/scenario/scenarios/airt/cyber.py @@ -23,12 +23,9 @@ logger = logging.getLogger(__name__) -# Techniques Cyber selects from the shared catalog. Cyber declares its own DEFAULT by naming -# these techniques (see _build_cyber_technique), rather than relying on a catalog-wide ``default`` -# tag. Adding a technique here that is not also in ``_CYBER_DEFAULT_TECHNIQUE_NAMES`` would keep -# it in ALL but out of DEFAULT, breaking the current DEFAULT == ALL invariant guarded by -# test_default_matches_all; keep the two sets in sync if this list grows. -_CYBER_TECHNIQUE_NAMES = {"red_teaming"} +# Cyber curates its DEFAULT run to the technique(s) named here (see _build_cyber_technique). +# The pool of *available* techniques is not narrowed — cyber exposes whatever the active +# initializer has registered (like RapidResponse); the initializer is the single gate. _CYBER_DEFAULT_TECHNIQUE_NAMES = {"red_teaming"} @@ -37,34 +34,26 @@ def _build_cyber_technique() -> type[ScenarioTechnique]: """ Build the Cyber technique class dynamically from the registered technique factories. - Selects only the ``red_teaming`` factory from the singleton - ``AttackTechniqueRegistry``. A plain ``PromptSendingAttack`` baseline is - prepended automatically by ``Scenario._build_baseline_atomic_attack`` via - ``BaselineAttackPolicy.Enabled``. + Exposes every technique registered in the singleton ``AttackTechniqueRegistry``; + which techniques are available is decided by the active initializer, not narrowed + here. A plain ``PromptSendingAttack`` baseline is prepended automatically by + ``Scenario._build_baseline_atomic_attack`` via ``BaselineAttackPolicy.Enabled``. - The ``DEFAULT`` aggregate is the curated default run; for Cyber it expands to the - same single ``red_teaming`` technique as ``ALL``. + The ``DEFAULT`` aggregate is the curated default run — for Cyber it expands to + ``red_teaming`` — while ``ALL`` selects the full registered pool. Returns: type[ScenarioTechnique]: The dynamically generated technique enum class. """ from pyrit.registry.components.attack_technique_registry import AttackTechniqueRegistry - from pyrit.registry.tag_query import TagQuery registry = AttackTechniqueRegistry.get_registry_singleton() - factories = registry.get_factories_or_raise() - cyber_factories = [f for name, f in factories.items() if name in _CYBER_TECHNIQUE_NAMES] + factories = list(registry.get_factories_or_raise().values()) return AttackTechniqueRegistry.build_technique_class_from_factories( # type: ignore[ty:invalid-return-type] class_name="CyberTechnique", - factories=cyber_factories, - aggregate_tags={ - "multi_turn": TagQuery.any_of("multi_turn"), - }, - # Cyber curates a single technique (red_teaming) at the scenario level. It declares that - # as its DEFAULT by name rather than tagging red_teaming ``default`` globally, which would - # alter other scenarios. - default_technique_names=_CYBER_DEFAULT_TECHNIQUE_NAMES, + factories=factories, + default_names=_CYBER_DEFAULT_TECHNIQUE_NAMES, ) @@ -77,7 +66,10 @@ class Cyber(Scenario): techniques. """ - VERSION: int = 2 + #: Bumped from 2 → 3 by dropping the ``core`` pool gate so the selectable + #: technique pool (and the ``all`` aggregate) reflects whatever the initializer + #: registered. ``use_cached`` only matches prior runs at the current ``VERSION``. + VERSION: int = 3 @classmethod def get_override_composite_scorer_questions_path(cls) -> list[Path]: @@ -114,7 +106,6 @@ def __init__( version=self.VERSION, objective_scorer=self._objective_scorer, technique_class=technique_class, - default_technique=technique_class("default"), default_dataset_config=DatasetAttackConfiguration(dataset_names=["airt_malware"], max_dataset_size=4), scenario_result_id=scenario_result_id, ) diff --git a/pyrit/scenario/scenarios/airt/jailbreak.py b/pyrit/scenario/scenarios/airt/jailbreak.py index 16157f0316..acd8ac356c 100644 --- a/pyrit/scenario/scenarios/airt/jailbreak.py +++ b/pyrit/scenario/scenarios/airt/jailbreak.py @@ -64,6 +64,11 @@ def get_aggregate_tags(cls) -> set[str]: # Include base class aggregates ("all") and add scenario-specific ones return super().get_aggregate_tags() | {"simple", "complex"} + @classmethod + def default(cls) -> "JailbreakTechnique": + """Return the default technique (``SIMPLE``) used when the caller selects nothing.""" + return cls.SIMPLE + class Jailbreak(Scenario): """ @@ -145,7 +150,6 @@ def __init__( super().__init__( version=self.VERSION, technique_class=JailbreakTechnique, - default_technique=JailbreakTechnique.SIMPLE, default_dataset_config=DatasetAttackConfiguration(dataset_names=["airt_harms"], max_dataset_size=4), objective_scorer=self._objective_scorer, scenario_result_id=scenario_result_id, diff --git a/pyrit/scenario/scenarios/airt/leakage.py b/pyrit/scenario/scenarios/airt/leakage.py index 0166a36564..7fb356cbc8 100644 --- a/pyrit/scenario/scenarios/airt/leakage.py +++ b/pyrit/scenario/scenarios/airt/leakage.py @@ -10,7 +10,6 @@ from pyrit.common import apply_defaults from pyrit.common.path import SCORER_SEED_PROMPT_PATH from pyrit.registry.components.attack_technique_registry import AttackTechniqueRegistry -from pyrit.registry.tag_query import TagQuery 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 @@ -67,11 +66,7 @@ def _build_leakage_technique() -> type[ScenarioTechnique]: return AttackTechniqueRegistry.build_technique_class_from_factories( # type: ignore[return-value, ty:invalid-return-type] class_name="LeakageTechnique", factories=all_factories, - aggregate_tags={ - "single_turn": TagQuery.any_of("single_turn"), - "multi_turn": TagQuery.any_of("multi_turn"), - }, - default_technique_names={"role_play_movie_script", "many_shot", "first_letter", "image"}, + default_names={"role_play_movie_script", "many_shot", "first_letter", "image"}, ) @@ -124,7 +119,6 @@ def __init__( super().__init__( version=self.VERSION, technique_class=technique_class, - default_technique=technique_class("default"), default_dataset_config=DatasetAttackConfiguration(dataset_names=["airt_leakage"], max_dataset_size=4), objective_scorer=objective_scorer, scenario_result_id=scenario_result_id, diff --git a/pyrit/scenario/scenarios/airt/psychosocial.py b/pyrit/scenario/scenarios/airt/psychosocial.py index e7f919b552..d324b1a43b 100644 --- a/pyrit/scenario/scenarios/airt/psychosocial.py +++ b/pyrit/scenario/scenarios/airt/psychosocial.py @@ -224,7 +224,6 @@ def __init__( super().__init__( version=self.VERSION, technique_class=PsychosocialTechnique, - default_technique=PsychosocialTechnique.ALL, default_dataset_config=DatasetAttackConfiguration( dataset_names=["airt_imminent_crisis"], max_dataset_size=4 ), diff --git a/pyrit/scenario/scenarios/airt/rapid_response.py b/pyrit/scenario/scenarios/airt/rapid_response.py index 8362d53595..c981657c48 100644 --- a/pyrit/scenario/scenarios/airt/rapid_response.py +++ b/pyrit/scenario/scenarios/airt/rapid_response.py @@ -35,14 +35,14 @@ def _build_rapid_response_technique() -> type[ScenarioTechnique]: """ Build the RapidResponse technique class dynamically from the registered factories. - Reads the singleton ``AttackTechniqueRegistry`` and filters to factories - tagged ``core``. + Reads every technique registered in the singleton ``AttackTechniqueRegistry`` + and exposes all of them. Which techniques are available is decided by the + active initializer (the registration gate), not narrowed again here. Returns: type[ScenarioTechnique]: The dynamically generated technique enum class. """ from pyrit.registry.components.attack_technique_registry import AttackTechniqueRegistry - from pyrit.registry.tag_query import TagQuery registry = AttackTechniqueRegistry.get_registry_singleton() factories = list(registry.get_factories_or_raise().values()) @@ -50,12 +50,7 @@ def _build_rapid_response_technique() -> type[ScenarioTechnique]: return AttackTechniqueRegistry.build_technique_class_from_factories( # type: ignore[ty:invalid-return-type] class_name="RapidResponseTechnique", factories=factories, - available=TagQuery.all("core"), - aggregate_tags={ - "single_turn": TagQuery.any_of("single_turn"), - "multi_turn": TagQuery.any_of("multi_turn"), - }, - default_technique_names={"role_play_movie_script", "many_shot"}, + default_tags={"light"}, ) @@ -67,7 +62,10 @@ class RapidResponse(Scenario): techniques. """ - VERSION: int = 2 + #: Bumped from 2 → 3 by dropping the ``core`` pool gate so the selectable + #: technique pool (and the ``all`` aggregate) reflects whatever the initializer + #: registered. ``use_cached`` only matches prior runs at the current ``VERSION``. + VERSION: int = 3 @apply_defaults def __init__( @@ -96,7 +94,6 @@ def __init__( version=self.VERSION, objective_scorer=self._objective_scorer, technique_class=technique_class, - default_technique=technique_class("default"), default_dataset_config=CompoundDatasetAttackConfiguration.per_dataset( dataset_names=[ "airt_hate", diff --git a/pyrit/scenario/scenarios/airt/scam.py b/pyrit/scenario/scenarios/airt/scam.py index a22248b960..804c0e3a07 100644 --- a/pyrit/scenario/scenarios/airt/scam.py +++ b/pyrit/scenario/scenarios/airt/scam.py @@ -75,6 +75,11 @@ def get_aggregate_tags(cls) -> set[str]: # Include base class aggregates ("all") and add scenario-specific ones return super().get_aggregate_tags() | {"default", "single_turn", "multi_turn"} + @classmethod + def default(cls) -> "ScamTechnique": + """Return the default technique (``DEFAULT``) used when the caller selects nothing.""" + return cls.DEFAULT + class Scam(Scenario): """ @@ -145,7 +150,6 @@ def __init__( super().__init__( version=self.VERSION, technique_class=ScamTechnique, - default_technique=ScamTechnique.DEFAULT, default_dataset_config=DatasetAttackConfiguration(dataset_names=["airt_scams"], max_dataset_size=4), objective_scorer=objective_scorer, scenario_result_id=scenario_result_id, diff --git a/pyrit/scenario/scenarios/benchmark/adversarial.py b/pyrit/scenario/scenarios/benchmark/adversarial.py index a62bda6a3b..c6b3201add 100644 --- a/pyrit/scenario/scenarios/benchmark/adversarial.py +++ b/pyrit/scenario/scenarios/benchmark/adversarial.py @@ -14,7 +14,6 @@ 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.scenario import BaselineAttackPolicy, Scenario @@ -35,15 +34,17 @@ def _build_benchmark_technique() -> type[ScenarioTechnique]: """ Build the ``BenchmarkTechnique`` enum from the registered factory catalog. - Reads ``core`` adversarial-capable factories from the + Reads adversarial-capable factories from the ``AttackTechniqueRegistry`` singleton and passes them to ``build_technique_class_from_factories``. Factories that bake their own ``adversarial_chat`` are excluded — the benchmark sweeps each technique across the user-supplied targets, which is incompatible with a technique - that pins its own adversarial target. The resulting enum has one + that pins its own adversarial target. Which techniques are registered is + decided by the active initializer (the registration gate); this scenario + does not narrow the pool further by group. The resulting enum has one concrete member per factory (e.g. ``red_teaming``, ``tap``, - ``crescendo_simulated``) plus ``default`` / ``light`` / ``single_turn`` - / ``multi_turn`` aggregates derived from each factory's ``technique_tags``. + ``crescendo_simulated``) and a ``light`` / ``single_turn`` / ``multi_turn`` + aggregate for each catalog tag. ``light`` is the scenario's default run. The (technique × target) cross-product is materialized lazily in ``AdversarialBenchmark._build_atomic_attacks_async`` from the @@ -56,17 +57,12 @@ def _build_benchmark_technique() -> type[ScenarioTechnique]: factories = [ factory for factory in registry.get_factories_or_raise().values() - if factory.uses_adversarial and "core" in factory.technique_tags and factory.adversarial_chat is None + if factory.uses_adversarial and factory.adversarial_chat is None ] return AttackTechniqueRegistry.build_technique_class_from_factories( # type: ignore[ty:invalid-return-type] class_name="BenchmarkTechnique", factories=factories, - aggregate_tags={ - "light": TagQuery.any_of("light"), - "single_turn": TagQuery.any_of("single_turn"), - "multi_turn": TagQuery.any_of("multi_turn"), - }, - default_technique_names={"role_play_movie_script", "many_shot"}, + default_tags={"light"}, ) @@ -82,7 +78,7 @@ class AdversarialBenchmark(Scenario): At run time, ``_build_atomic_attacks_async`` performs the ``(technique × adversarial_target × dataset)`` cross-product: for each - selected adversarial-capable ``core`` factory in the + selected adversarial-capable factory in the ``AttackTechniqueRegistry`` and each requested target, it calls ``factory.create(adversarial_chat=...)`` with the resolved target — no global registry mutation. The resulting @@ -95,9 +91,12 @@ class AdversarialBenchmark(Scenario): #: from a constructor parameter to the ``adversarial_targets`` scenario #: parameter and changed ``atomic_attack_name`` from #: ``{technique}__{model}__{dataset}`` to ``{technique}__{target}_{dataset}``. + #: Bumped from 2 → 3 by dropping the ``core`` pool gate so the selectable + #: technique pool (and therefore the ``all`` aggregate) reflects whatever the + #: initializer registered rather than only core-tagged factories. #: ``use_cached`` only matches against prior runs at the current - #: ``VERSION``; v1 results remain queryable but won't suppress v2 runs. - VERSION: int = 2 + #: ``VERSION``; older results remain queryable but won't suppress v3 runs. + VERSION: int = 3 #: AdversarialBenchmark compares attack-success rates across adversarial models; a baseline #: attack would be model-independent and contribute no signal to the comparison. @@ -181,7 +180,6 @@ def __init__( version=self.VERSION, objective_scorer=self._objective_scorer, technique_class=technique_class, - default_technique=technique_class("light"), default_dataset_config=DatasetAttackConfiguration( dataset_names=["harmbench"], max_dataset_size=8, diff --git a/pyrit/scenario/scenarios/foundry/red_team_agent.py b/pyrit/scenario/scenarios/foundry/red_team_agent.py index cee05fbc31..d05b8e53fd 100644 --- a/pyrit/scenario/scenarios/foundry/red_team_agent.py +++ b/pyrit/scenario/scenarios/foundry/red_team_agent.py @@ -186,6 +186,11 @@ def get_aggregate_tags(cls) -> set[str]: # Include base class aggregates ("all") and add Foundry-specific ones return super().get_aggregate_tags() | {"easy", "moderate", "difficult", "converter", "attack"} + @classmethod + def default(cls) -> "FoundryTechnique": + """Return the default technique (``EASY``) used when the caller selects nothing.""" + return cls.EASY + class RedTeamAgent(Scenario): """ @@ -242,7 +247,6 @@ def __init__( super().__init__( version=self.VERSION, technique_class=FoundryTechnique, - default_technique=FoundryTechnique.EASY, default_dataset_config=DatasetAttackConfiguration(dataset_names=["harmbench"], max_dataset_size=4), objective_scorer=objective_scorer, scenario_result_id=scenario_result_id, diff --git a/pyrit/scenario/scenarios/garak/doctor.py b/pyrit/scenario/scenarios/garak/doctor.py index 9c45174471..5daa89c95b 100644 --- a/pyrit/scenario/scenarios/garak/doctor.py +++ b/pyrit/scenario/scenarios/garak/doctor.py @@ -82,8 +82,7 @@ def _build_doctor_technique() -> type[ScenarioTechnique]: return AttackTechniqueRegistry.build_technique_class_from_factories( # type: ignore[return-value, ty:invalid-return-type] class_name="DoctorTechnique", factories=DOCTOR_FACTORIES, - aggregate_tags={}, - default_technique_names={"policy_puppetry", "policy_puppetry_leet"}, + default_names={"policy_puppetry", "policy_puppetry_leet"}, ) @@ -138,7 +137,6 @@ def __init__( super().__init__( version=self.VERSION, technique_class=technique_class, - default_technique=technique_class("default"), default_dataset_config=DatasetAttackConfiguration(dataset_names=["garak_doctor"]), objective_scorer=objective_scorer, scenario_result_id=scenario_result_id, diff --git a/pyrit/scenario/scenarios/garak/encoding.py b/pyrit/scenario/scenarios/garak/encoding.py index e4b23e13ac..d9e409b671 100644 --- a/pyrit/scenario/scenarios/garak/encoding.py +++ b/pyrit/scenario/scenarios/garak/encoding.py @@ -122,6 +122,16 @@ def get_aggregate_tags(cls) -> set[str]: """ return super().get_aggregate_tags() | {"default"} + @classmethod + def default(cls) -> "EncodingTechnique": + """ + Select the curated ``DEFAULT`` aggregate for the out-of-the-box run, not the exhaustive ``ALL``. + + Returns: + EncodingTechnique: The curated ``DEFAULT`` aggregate. + """ + return cls.DEFAULT + logger = logging.getLogger(__name__) @@ -172,7 +182,6 @@ def __init__( super().__init__( version=self.VERSION, technique_class=EncodingTechnique, - default_technique=EncodingTechnique.DEFAULT, default_dataset_config=CompoundDatasetAttackConfiguration( configurations=[ EncodingDatasetConfiguration(dataset_names=["garak_slur_terms_en"], max_dataset_size=10), diff --git a/pyrit/scenario/scenarios/garak/web_injection.py b/pyrit/scenario/scenarios/garak/web_injection.py index 962837a892..dc29e7117e 100644 --- a/pyrit/scenario/scenarios/garak/web_injection.py +++ b/pyrit/scenario/scenarios/garak/web_injection.py @@ -198,6 +198,11 @@ def get_aggregate_tags(cls) -> set[str]: """Return the tags that represent aggregate categories.""" return {"all", "default", "exfil", "xss"} + @classmethod + def default(cls) -> WebInjectionTechnique: + """Return the default technique (``DEFAULT``) used when the caller selects nothing.""" + return cls.DEFAULT + class WebInjection(Scenario): """ @@ -260,7 +265,6 @@ def __init__( super().__init__( version=self.VERSION, technique_class=WebInjectionTechnique, - default_technique=WebInjectionTechnique.DEFAULT, default_dataset_config=DatasetAttackConfiguration( dataset_names=[ DATASET_EXAMPLE_DOMAINS, diff --git a/pyrit/setup/initializers/techniques/core.py b/pyrit/setup/initializers/techniques/core.py index 366f160d88..2b26777cb8 100644 --- a/pyrit/setup/initializers/techniques/core.py +++ b/pyrit/setup/initializers/techniques/core.py @@ -12,7 +12,7 @@ ``default`` is intentionally not a tag here: what runs by default is scenario-relative and is declared per scenario (see ``AttackTechniqueRegistry.build_technique_class_from_factories``'s -``default_technique_names``), not baked into the shared catalog. +``default_tags`` / ``default_names``), not baked into the shared catalog. """ from pyrit.common.path import ( diff --git a/tests/unit/registry/test_attack_technique_registry.py b/tests/unit/registry/test_attack_technique_registry.py index d0a109d365..ebff9d3ef7 100644 --- a/tests/unit/registry/test_attack_technique_registry.py +++ b/tests/unit/registry/test_attack_technique_registry.py @@ -578,3 +578,76 @@ def test_factory_skips_silently_when_attack_has_no_scoring_param_and_policy_skip assert technique is not None assert not any("does not accept" in record.message for record in caplog.records) + + +def _make_pool() -> list[AttackTechniqueFactory]: + """A small pool of tagged factories for exercising the technique-class builder.""" + return [ + AttackTechniqueFactory(name="alpha", attack_class=_StubAttack, technique_tags=["light", "single_turn"]), + AttackTechniqueFactory(name="beta", attack_class=_StubAttack, technique_tags=["light", "multi_turn"]), + AttackTechniqueFactory(name="gamma", attack_class=_StubAttack, technique_tags=["multi_turn"]), + ] + + +class TestBuildTechniqueClassFromFactories: + """Tests for the ``build_technique_class_from_factories`` default/aggregate contract.""" + + def test_catalog_tags_auto_promote_to_aggregates(self): + """Every catalog tag in the pool becomes a selectable aggregate expanding to its techniques.""" + cls = AttackTechniqueRegistry.build_technique_class_from_factories( + class_name="AutoAggTechnique", + factories=_make_pool(), + ) + aggregates = cls.get_aggregate_tags() + assert {"all", "light", "single_turn", "multi_turn"} <= aggregates + assert {c.value for c in cls.expand({cls("light")})} == {"alpha", "beta"} + + def test_default_tags_builds_default_aggregate_and_default_returns_it(self): + """``default_tags`` builds the DEFAULT aggregate; ``default()`` returns it, expanding to tagged techniques.""" + cls = AttackTechniqueRegistry.build_technique_class_from_factories( + class_name="DefaultTagsTechnique", + factories=_make_pool(), + default_tags={"light"}, + ) + assert cls.default() == cls("default") + assert {c.value for c in cls.expand({cls.default()})} == {"alpha", "beta"} + + def test_default_names_builds_default_aggregate_from_names(self): + """``default_names`` builds the DEFAULT aggregate from exact names.""" + cls = AttackTechniqueRegistry.build_technique_class_from_factories( + class_name="DefaultNamesTechnique", + factories=_make_pool(), + default_names={"gamma"}, + ) + assert cls.default() == cls("default") + assert {c.value for c in cls.expand({cls.default()})} == {"gamma"} + + def test_no_default_leaves_attribute_unset_and_falls_back_to_all(self): + """With no default, the builder records nothing and ``default()`` owns the single ALL fallback.""" + cls = AttackTechniqueRegistry.build_technique_class_from_factories( + class_name="NoDefaultTechnique", + factories=_make_pool(), + ) + assert not hasattr(cls, "_default_technique_value") + assert "default" not in cls.get_aggregate_tags() + assert cls.default() == cls["ALL"] + + def test_default_selection_matching_nothing_falls_back_to_all(self): + """A default set that matches no pool technique builds no DEFAULT and falls back to ALL.""" + cls = AttackTechniqueRegistry.build_technique_class_from_factories( + class_name="EmptyDefaultTechnique", + factories=_make_pool(), + default_tags={"nonexistent"}, + ) + assert not hasattr(cls, "_default_technique_value") + assert cls.default() == cls["ALL"] + + def test_both_default_tags_and_names_raises(self): + """``default_tags`` and ``default_names`` are mutually exclusive.""" + with pytest.raises(ValueError, match="at most one of default_tags or default_names"): + AttackTechniqueRegistry.build_technique_class_from_factories( + class_name="BothDefaultsTechnique", + factories=_make_pool(), + default_tags={"light"}, + default_names={"gamma"}, + ) diff --git a/tests/unit/scenario/airt/test_cyber.py b/tests/unit/scenario/airt/test_cyber.py index 70ff711f10..d29d3b5caa 100644 --- a/tests/unit/scenario/airt/test_cyber.py +++ b/tests/unit/scenario/airt/test_cyber.py @@ -126,8 +126,8 @@ def _make_seed_groups(name: str) -> list[AttackSeedGroup]: class TestCyberBasic: """Tests for Cyber initialization and class properties.""" - def test_version_is_2(self): - assert Cyber.VERSION == 2 + def test_version_is_3(self): + assert Cyber.VERSION == 3 def test_get_technique_class(self): strat = _technique_class() @@ -149,15 +149,17 @@ def test_default_aggregate_expands_to_red_teaming(self): default_members = strat.expand({strat.DEFAULT}) assert default_members == [strat("red_teaming")] - def test_default_matches_all(self): - """DEFAULT must expand to exactly the same techniques as ALL. + def test_all_is_a_superset_of_default(self): + """ALL exposes the full registered pool while DEFAULT stays curated to red_teaming. - Cyber curates a single technique, so its DEFAULT run is a no-op alias of ALL. Asserting - equality (not just subset) guards against a future technique landing in ALL but being - silently excluded from DEFAULT (or vice versa) once the aggregate wiring changes. + Cyber no longer narrows its available techniques — the initializer is the single + gate — so ALL is strictly broader than the curated DEFAULT run. """ strat = _technique_class() - assert set(strat.expand({strat.DEFAULT})) == set(strat.expand({strat.ALL})) + default_members = set(strat.expand({strat.DEFAULT})) + all_members = set(strat.expand({strat.ALL})) + assert default_members == {strat("red_teaming")} + assert default_members < all_members def test_default_dataset_config_has_malware_dataset(self): config = Cyber()._default_dataset_config @@ -296,23 +298,29 @@ async def _init_and_get_attacks( await scenario.initialize_async() return scenario._atomic_attacks - async def test_all_technique_produces_red_teaming(self, mock_objective_target, mock_objective_scorer): + async def test_all_technique_includes_red_teaming_and_others(self, mock_objective_target, mock_objective_scorer): + """ALL now exposes the full registered pool, not just red_teaming.""" attacks = await self._init_and_get_attacks( mock_objective_target=mock_objective_target, mock_objective_scorer=mock_objective_scorer, techniques=[_technique_class().ALL], ) technique_classes = {type(a.attack_technique.attack) for a in attacks} - assert technique_classes == {RedTeamingAttack} + assert RedTeamingAttack in technique_classes + assert len(technique_classes) > 1 - async def test_multi_turn_technique_produces_red_teaming(self, mock_objective_target, mock_objective_scorer): + async def test_multi_turn_technique_includes_red_teaming_and_others( + self, mock_objective_target, mock_objective_scorer + ): + """MULTI_TURN exposes every registered multi-turn technique, including red_teaming.""" attacks = await self._init_and_get_attacks( mock_objective_target=mock_objective_target, mock_objective_scorer=mock_objective_scorer, techniques=[_technique_class().MULTI_TURN], ) technique_classes = {type(a.attack_technique.attack) for a in attacks} - assert technique_classes == {RedTeamingAttack} + assert RedTeamingAttack in technique_classes + assert len(technique_classes) > 1 async def test_default_technique_produces_red_teaming(self, mock_objective_target, mock_objective_scorer): """Default (DEFAULT) should produce RedTeaming. PromptSendingAttack baseline is diff --git a/tests/unit/scenario/airt/test_rapid_response.py b/tests/unit/scenario/airt/test_rapid_response.py index 9a974b4b91..66259563e0 100644 --- a/tests/unit/scenario/airt/test_rapid_response.py +++ b/tests/unit/scenario/airt/test_rapid_response.py @@ -155,8 +155,8 @@ def _make_seed_groups(name: str) -> list[AttackSeedGroup]: class TestRapidResponseBasic: """Tests for RapidResponse initialization and class properties.""" - def test_version_is_2(self): - assert RapidResponse.VERSION == 2 + def test_version_is_3(self): + assert RapidResponse.VERSION == 3 def test_get_technique_class(self, mock_objective_scorer): strat = _technique_class() @@ -170,7 +170,10 @@ def test_get_default_technique_returns_default(self, mock_objective_scorer): with patch( "pyrit.scenario.core.scenario.Scenario._get_default_objective_scorer", return_value=mock_objective_scorer ): - assert RapidResponse()._default_technique == strat.DEFAULT + default = RapidResponse()._default_technique + assert default == strat.DEFAULT + # DEFAULT is built from default_tags={"light"}, so it expands to the light-tagged techniques. + assert strat.expand({strat.DEFAULT}) == strat.expand({strat.LIGHT}) def test_default_dataset_config_has_all_harm_datasets(self, mock_objective_scorer): with patch( @@ -210,7 +213,7 @@ def test_initialization_with_custom_scorer(self, mock_objective_scorer): new_callable=AsyncMock, return_value=ALL_HARM_SEED_GROUPS, ) - async def test_initialization_defaults_to_default_technique( + async def test_initialization_defaults_to_light_technique( self, _mock_groups, mock_get_scorer, @@ -221,8 +224,11 @@ async def test_initialization_defaults_to_default_technique( scenario = RapidResponse() scenario.set_params_from_args(args={"objective_target": mock_objective_target}) await scenario.initialize_async() - # DEFAULT expands to PromptSending + ManyShot → 2 composites - assert len(scenario._scenario_techniques) == 2 + # Default is the DEFAULT aggregate (built from light tags); it expands to every light-tagged technique. + strat = _technique_class() + expected = len(strat.expand({strat.DEFAULT})) + assert expected > 2 + assert len(scenario._scenario_techniques) == expected async def test_initialize_raises_when_no_datasets(self, mock_objective_target, mock_objective_scorer): """Dataset resolution fails from empty memory.""" @@ -306,16 +312,18 @@ async def _init_and_get_attacks( await scenario.initialize_async() return scenario._atomic_attacks - async def test_default_technique_produces_role_play_and_many_shot( - self, mock_objective_target, mock_objective_scorer - ): + async def test_default_technique_is_light(self, mock_objective_target, mock_objective_scorer): attacks = await self._init_and_get_attacks( mock_objective_target=mock_objective_target, mock_objective_scorer=mock_objective_scorer, ) technique_classes = {type(a.attack_technique.attack) for a in attacks} - # role_play_movie_script is now a simulated-conversation PromptSendingAttack. + # Default is the DEFAULT aggregate (built from light tags): it includes many_shot and the + # simulated-conversation techniques (context_compliance, role_play_*), but excludes the slow + # TAP attack. context_compliance and role_play_* are now simulated-conversation + # PromptSendingAttacks, so assert on the simulated-conversation seed rather than a dedicated class. assert ManyShotJailbreakAttack in technique_classes + assert TreeOfAttacksWithPruningAttack not in technique_classes assert any( a.attack_technique.seed_technique is not None and a.attack_technique.seed_technique.has_simulated_conversation @@ -432,7 +440,7 @@ def _spy_create(self, **kwargs): assert len(extra) == 1 async def test_attack_count_is_techniques_times_datasets(self, mock_objective_target, mock_objective_scorer): - """With 2 datasets and DEFAULT (2 techniques), expect 4 atomic attacks.""" + """With 2 datasets and the DEFAULT (light) default, expect (light techniques) x 2 atomic attacks.""" two_datasets = { "hate": _make_seed_groups("hate"), "violence": _make_seed_groups("violence"), @@ -442,8 +450,9 @@ async def test_attack_count_is_techniques_times_datasets(self, mock_objective_ta mock_objective_scorer=mock_objective_scorer, seed_groups=two_datasets, ) - # DEFAULT = RolePlay + ManyShot = 2 techniques, 2 datasets → 4 - assert len(attacks) == 4 + strat = _technique_class() + expected = len(strat.expand({strat.DEFAULT})) * 2 + assert len(attacks) == expected async def test_atomic_attack_names_are_unique_compound_keys(self, mock_objective_target, mock_objective_scorer): """Each AtomicAttack has a unique compound atomic_attack_name for resume correctness.""" diff --git a/tests/unit/scenario/benchmark/test_adversarial.py b/tests/unit/scenario/benchmark/test_adversarial.py index 8b1cbf2ecc..6edcc0dda9 100644 --- a/tests/unit/scenario/benchmark/test_adversarial.py +++ b/tests/unit/scenario/benchmark/test_adversarial.py @@ -15,7 +15,7 @@ These tests cover the new contract: * Class metadata (VERSION, BASELINE policy, defaults). * Technique enum is built from registered factories with ``uses_adversarial=True`` - and the ``core`` technique tag; ``light`` aggregate preserves the + that do not bake their own ``adversarial_chat``; ``light`` aggregate preserves the source ``light`` tag (excludes ``tap`` / ``crescendo_simulated``). * ``supported_parameters`` declares ``adversarial_targets: list[str]``. * ``_resolve_adversarial_targets`` raises with available names on typos. @@ -76,7 +76,7 @@ def _build_benchmarkable_factories_snapshot() -> list: factories = build_technique_factories() finally: TargetRegistry.reset_registry_singleton() - return [f for f in factories if f.uses_adversarial and "core" in f.technique_tags] + return [f for f in factories if f.uses_adversarial and f.adversarial_chat is None] _BENCHMARKABLE_FACTORIES = _build_benchmarkable_factories_snapshot() @@ -126,6 +126,7 @@ def _register_mock_factory(*, name: str, tags: list[str] | None = None, seed_tec factory = MagicMock(spec=AttackTechniqueFactory) factory.name = name factory.uses_adversarial = True + factory.adversarial_chat = None factory.technique_tags = tags if tags is not None else ["core", "light"] factory.seed_technique = seed_technique technique_instance = MagicMock(name="AttackTechnique") @@ -154,9 +155,9 @@ async def _build_atomic_attacks(bench: AdversarialBenchmark) -> list: class TestAdversarialBenchmarkMetadata: """Tests for class-level metadata that doesn't depend on any runtime state.""" - def test_version_is_2(self): - """VERSION matches the post-collapse ``atomic_attack_name`` format so cached results still match.""" - assert AdversarialBenchmark.VERSION == 2 + def test_version_is_3(self): + """VERSION bumped to 3 when the ``core`` pool gate was dropped so cached v2 results don't suppress v3 runs.""" + assert AdversarialBenchmark.VERSION == 3 def test_baseline_attack_policy_is_forbidden(self): """A baseline contributes no signal to a model-comparison benchmark, so it is forbidden.""" @@ -202,7 +203,7 @@ class TestAdversarialBenchmarkTechnique: """Tests for ``_build_benchmark_technique`` using the registry-based factory API.""" def test_technique_built_from_registered_adversarial_factories(self): - """Each registered ``core`` adversarial factory produces one concrete enum member.""" + """Each registered adversarial factory produces one concrete enum member.""" technique_cls = _build_benchmark_technique() aggregate_names = {"all"} | technique_cls.get_aggregate_tags() concrete_members = [m for m in technique_cls if m.value not in aggregate_names] @@ -240,6 +241,8 @@ def test_technique_excludes_factories_with_baked_adversarial_chat(self): technique_cls = _build_benchmark_technique() member_values = {m.value for m in technique_cls} assert "pinned_adversary" not in member_values + + def test_technique_exposes_tag_aggregates(self): """The technique enum exposes ``light``, ``single_turn``, ``multi_turn`` aggregates.""" technique_cls = _build_benchmark_technique() aggregates = technique_cls.get_aggregate_tags() @@ -305,6 +308,17 @@ def test_skip_cached_can_be_set_true(self): ) assert bench._use_cached is True + def test_construct_without_light_factory_falls_back_to_all(self): + """A pool with no ``light``-tagged factory must still construct, defaulting to ``all``.""" + AttackTechniqueRegistry.reset_registry_singleton() + _build_benchmark_technique.cache_clear() + _register_mock_factory(name="narrow_tap", tags=["airt_internal", "multi_turn"]) + + bench = AdversarialBenchmark(objective_scorer=MagicMock(spec=TrueFalseScorer)) + + assert "light" not in bench._technique_class.get_aggregate_tags() + assert bench._default_technique.value == "all" + # --------------------------------------------------------------------------- # _resolve_adversarial_targets diff --git a/tests/unit/scenario/core/test_scenario.py b/tests/unit/scenario/core/test_scenario.py index 8bf9b368ae..b496c86718 100644 --- a/tests/unit/scenario/core/test_scenario.py +++ b/tests/unit/scenario/core/test_scenario.py @@ -156,7 +156,6 @@ def get_aggregate_tags(cls) -> set[str]: return {"all"} kwargs.setdefault("technique_class", TestTechnique) - kwargs.setdefault("default_technique", kwargs["technique_class"].ALL) kwargs.setdefault("default_dataset_config", DatasetConfiguration()) # Add a mock scorer if not provided @@ -772,7 +771,6 @@ def get_aggregate_tags(cls) -> set[str]: return {"all"} kwargs.setdefault("technique_class", TestTechnique) - kwargs.setdefault("default_technique", kwargs["technique_class"].ALL) kwargs.setdefault("default_dataset_config", DatasetConfiguration()) # Use TrueFalseScorer mock if not provided diff --git a/tests/unit/scenario/core/test_scenario_parameters.py b/tests/unit/scenario/core/test_scenario_parameters.py index b7512e4df3..98fe9e28cb 100644 --- a/tests/unit/scenario/core/test_scenario_parameters.py +++ b/tests/unit/scenario/core/test_scenario_parameters.py @@ -65,7 +65,6 @@ async def _build_atomic_attacks_async(self, *, context): return _ParamTestScenario( version=1, technique_class=_ParamTestTechnique, - default_technique=_ParamTestTechnique.ALL, default_dataset_config=DatasetConfiguration(), objective_scorer=mock_scorer, ) diff --git a/tests/unit/scenario/core/test_scenario_partial_results.py b/tests/unit/scenario/core/test_scenario_partial_results.py index 74542813fe..c95fa7f2e6 100644 --- a/tests/unit/scenario/core/test_scenario_partial_results.py +++ b/tests/unit/scenario/core/test_scenario_partial_results.py @@ -104,7 +104,6 @@ def __init__(self, *, atomic_attacks_to_return=None, objective_scorer=None, **kw objective_scorer = MagicMock() objective_scorer.get_identifier.return_value = _mock_scorer_id("MockScorer") - kwargs.setdefault("default_technique", technique_class.ALL) kwargs.setdefault("default_dataset_config", DatasetConfiguration()) super().__init__(technique_class=technique_class, objective_scorer=objective_scorer, **kwargs) self._test_atomic_attacks = atomic_attacks_to_return or [] diff --git a/tests/unit/scenario/core/test_scenario_retry.py b/tests/unit/scenario/core/test_scenario_retry.py index 64553dc1dc..efd3651939 100644 --- a/tests/unit/scenario/core/test_scenario_retry.py +++ b/tests/unit/scenario/core/test_scenario_retry.py @@ -175,7 +175,6 @@ def __init__(self, *, atomic_attacks_to_return=None, objective_scorer=None, **kw objective_scorer = MagicMock() objective_scorer.get_identifier.return_value = _mock_scorer_id("MockScorer") - kwargs.setdefault("default_technique", technique_class.ALL) kwargs.setdefault("default_dataset_config", DatasetConfiguration()) super().__init__(technique_class=technique_class, objective_scorer=objective_scorer, **kwargs) self._atomic_attacks_to_return = atomic_attacks_to_return or [] diff --git a/tests/unit/scenario/core/test_scenario_technique_invariants.py b/tests/unit/scenario/core/test_scenario_technique_invariants.py index 9d0098a39d..dccbffa678 100644 --- a/tests/unit/scenario/core/test_scenario_technique_invariants.py +++ b/tests/unit/scenario/core/test_scenario_technique_invariants.py @@ -183,6 +183,27 @@ def test_aggregates_are_disjoint_from_techniques(get_technique): assert agg_values.isdisjoint(tech_values) +@pytest.mark.parametrize("get_technique", SCENARIO_STRATEGY_BUILDERS) +def test_catalog_tags_are_selectable_aggregates(get_technique): + """Every catalog tag carried by a pool technique is itself a selectable aggregate + that expands to exactly the techniques carrying it (tags are synonymous with + aggregates). ``all`` and ``default`` are reserved synthetic aggregates, and a tag + that collides with a technique name stays a concrete technique, so both are excluded. + """ + strat = get_technique() + reserved = {"all", "default"} + technique_values = {s.value for s in strat.get_all_techniques()} + catalog_tags = {tag for s in strat.get_all_techniques() for tag in s.tags} - reserved - technique_values + assert catalog_tags, "expected at least one catalog tag to promote" + + aggregate_values = {s.value for s in strat.get_aggregate_techniques()} + for tag in catalog_tags: + assert tag in aggregate_values, f"catalog tag {tag!r} is not a selectable aggregate" + expanded = set(strat.expand({strat(tag)})) + expected = {s for s in strat.get_all_techniques() if tag in s.tags} + assert expanded == expected, f"aggregate {tag!r} did not expand to its tagged techniques" + + @pytest.mark.parametrize("get_technique", SCENARIO_STRATEGY_BUILDERS) def test_expanding_a_technique_returns_itself(get_technique): """Expanding a single non-aggregate technique returns just that technique.""" diff --git a/tests/unit/scenario/scenarios/adaptive/test_text_adaptive.py b/tests/unit/scenario/scenarios/adaptive/test_text_adaptive.py index 383d5e4fd3..3efc712c86 100644 --- a/tests/unit/scenario/scenarios/adaptive/test_text_adaptive.py +++ b/tests/unit/scenario/scenarios/adaptive/test_text_adaptive.py @@ -135,9 +135,10 @@ def test_get_technique_class_is_cached(self): assert cls_a is cls_b def test_get_default_technique(self): - strat = TextAdaptive.get_default_technique() + strat = TextAdaptive.get_technique_class().default() # The default aggregate must resolve to something runnable. assert strat is not None + assert strat.value == "default" @patch("pyrit.scenario.core.scenario.Scenario._get_default_objective_scorer") def test_init_stores_adaptive_params(self, mock_get_scorer, mock_objective_scorer):