From 89ce00d0d4d4de77c44e531e2ba73075d116d62c Mon Sep 17 00:00:00 2001 From: Richard Lundeen Date: Thu, 9 Jul 2026 12:56:03 -0700 Subject: [PATCH 1/2] MAINT: Technique Organization Reorganize how attack techniques are cataloged, tagged, and selected so the catalog can scale to hundreds of techniques. Decisions: - Catalog lives under pyrit/setup/initializers/techniques/. `core.py` is a small, curated standard set; `extra.py` is the broader opt-in collection and the default home for new general-purpose contributions. Per-source modules (e.g. `airt.py`) hold techniques owned by a scenario/source but still reusable, tagged with their owner and kept out of the default pool. - Retire the global `default` tag. What runs by default is scenario-relative, so each scenario declares its own defaults via build_strategy_class_from_factories (available selects the pool, aggregates are named presets, default/default_technique_names set what runs when nothing is chosen). available/aggregate/default relate as strict subsets. - Techniques pinned to one scenario may stay local to it (e.g. Doctor's Policy Puppetry factories); promote to a catalog module only when reusable. - Migrate Doctor's hand-written strategy enum onto the shared factory generator (technique enum attr names go PascalCase -> snake_case; values unchanged). Default loading: - A bare initialize_pyrit_async now auto-runs TechniqueInitializer (core) and TargetInitializer when none are supplied, so `core` techniques and default targets are registered without any config. Users select other subsets by passing initializer tags (core/extra/all) or their own initializer, including from the CLI. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- doc/code/framework.md | 10 ++- .../components/attack_technique_registry.py | 67 ++++++++++++++++--- .../scenarios/adaptive/text_adaptive.py | 2 +- pyrit/scenario/scenarios/airt/cyber.py | 20 +++--- pyrit/scenario/scenarios/airt/leakage.py | 56 ++++++---------- .../scenario/scenarios/airt/rapid_response.py | 5 +- .../scenarios/benchmark/adversarial.py | 2 +- pyrit/scenario/scenarios/garak/doctor.py | 60 ++++++++--------- pyrit/setup/initialization.py | 24 ++++++- pyrit/setup/initializers/techniques/airt.py | 62 +++++++++++++++++ pyrit/setup/initializers/techniques/core.py | 45 +++++++------ tests/unit/scenario/garak/test_doctor.py | 4 +- 12 files changed, 241 insertions(+), 116 deletions(-) create mode 100644 pyrit/setup/initializers/techniques/airt.py diff --git a/doc/code/framework.md b/doc/code/framework.md index 3f646e4138..be9ff0a971 100644 --- a/doc/code/framework.md +++ b/doc/code/framework.md @@ -172,15 +172,19 @@ If you are contributing to PyRIT, that work will most likely land in one of the ## [Attack Techniques](./scenarios/0_attack_techniques) -**Responsibility**: An attack technique packages an executor, converters, datasets, and strategies into a single attack. The goal is that any attack (something trying to achieve an objective) can be defined as an attack technique. +**Responsibility**: An attack technique packages an executor, converters, datasets, and strategies into a single named attack. The goal is that any attack (something trying to achieve an objective) can be defined as an attack technique. +- Techniques are self-describing `AttackTechniqueFactory` instances (a `name`, `attack_class`, `attack_kwargs`, and `strategy_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_strategy_class_from_factories` (`available` selects the pool, aggregates are named presets, `default` / `default_technique_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**: -- Managing these better, so scenarios can more easily select or build the attack techniques to use +- Growing `extra` toward hundreds of techniques while keeping `core` small and curated, and making it easier to select technique subsets (via tags, initializers, or the CLI) without slow, noisy default runs. -**Contributing (difficulty easy)**: Simply add the attack technique to one of the initializers. +**Contributing (difficulty easy)**: Add an `AttackTechniqueFactory` to `extra.py` (the default home for new general-purpose techniques), or to a source-owned module (like `airt.py`) if it belongs to a specific scenario but could be reused. `core` is reserved for the curated standard set. Tag it with its behavioral tags; don't tag it `default`. ## [Executors and Attacks](./executor/0_executor) diff --git a/pyrit/registry/components/attack_technique_registry.py b/pyrit/registry/components/attack_technique_registry.py index 2d3ca12c24..ae5d99376f 100644 --- a/pyrit/registry/components/attack_technique_registry.py +++ b/pyrit/registry/components/attack_technique_registry.py @@ -178,44 +178,91 @@ def build_strategy_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, ) -> type: """ Build a ``ScenarioStrategy`` 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 factory, with tags from the factory. - - Each aggregate maps to a ``TagQuery`` that determines which - technique factories belong to it. + - 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. Args: class_name (str): Name for the generated enum class. - factories (list[AttackTechniqueFactory]): Technique factories to include - as enum members. + 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 techniques belong to the aggregate. - An ``ALL`` aggregate (expanding to all techniques) is always added. + ``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. Returns: type: A ``ScenarioStrategy`` subclass with the generated members. """ from pyrit.scenario import ScenarioStrategy + # available (the pool): filter the candidate factories by the availability query. + pool = available.filter(factories) if available is not None else list(factories) + + # 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.strategy_tags))} + all_aggregate_tag_names = {"all"} | set(aggregate_tags.keys()) + if default_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: + members["DEFAULT"] = ("default", {"default"}) for agg_name in aggregate_tags: members[agg_name.upper()] = (agg_name, {agg_name}) - # Technique members from factories — assign aggregate tags based on TagQuery matching - for factory in factories: + # Technique members from the pool — assign aggregate tags based on TagQuery matching + for factory in pool: factory_tags = set(factory.strategy_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) # Build the enum class dynamically diff --git a/pyrit/scenario/scenarios/adaptive/text_adaptive.py b/pyrit/scenario/scenarios/adaptive/text_adaptive.py index 7d1824ce52..22c3df0a73 100644 --- a/pyrit/scenario/scenarios/adaptive/text_adaptive.py +++ b/pyrit/scenario/scenarios/adaptive/text_adaptive.py @@ -72,10 +72,10 @@ def _build_text_adaptive_strategy() -> type[ScenarioStrategy]: class_name="TextAdaptiveStrategy", factories=factories, aggregate_tags={ - "default": TagQuery.any_of("default"), "single_turn": TagQuery.any_of("single_turn"), "multi_turn": TagQuery.any_of("multi_turn"), }, + default_technique_names={"role_play", "many_shot"}, ) diff --git a/pyrit/scenario/scenarios/airt/cyber.py b/pyrit/scenario/scenarios/airt/cyber.py index 079d905a6e..d9a2b12b1b 100644 --- a/pyrit/scenario/scenarios/airt/cyber.py +++ b/pyrit/scenario/scenarios/airt/cyber.py @@ -23,12 +23,13 @@ logger = logging.getLogger(__name__) -# Techniques Cyber selects from the shared catalog. ``DEFAULT`` is wired to ``any_of("core")`` -# (see _build_cyber_strategy), so adding a technique here that carries the ``core`` tag pulls it -# into DEFAULT, while a technique lacking ``core`` (e.g. an ``extra``-group technique) would stay -# in ALL but be silently dropped from DEFAULT. Either case breaks the current DEFAULT == ALL -# invariant (guarded by test_default_matches_all); revisit the aggregate wiring if that happens. +# Techniques Cyber selects from the shared catalog. Cyber declares its own DEFAULT by naming +# these techniques (see _build_cyber_strategy), 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_DEFAULT_TECHNIQUE_NAMES = {"red_teaming"} @cache @@ -58,13 +59,12 @@ def _build_cyber_strategy() -> type[ScenarioStrategy]: class_name="CyberStrategy", factories=cyber_factories, aggregate_tags={ - # Cyber curates a single technique (red_teaming) at the scenario level. That - # technique carries the canonical ``core`` tag but not the catalog-wide - # ``default`` tag, so DEFAULT matches ``core`` here to select it (rather than - # tagging red_teaming ``default`` globally, which would alter other scenarios). - "default": TagQuery.any_of("core"), "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, ) diff --git a/pyrit/scenario/scenarios/airt/leakage.py b/pyrit/scenario/scenarios/airt/leakage.py index 4346624a65..095687dc51 100644 --- a/pyrit/scenario/scenarios/airt/leakage.py +++ b/pyrit/scenario/scenarios/airt/leakage.py @@ -8,13 +8,9 @@ from typing import TYPE_CHECKING 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.prompt_converter import AddImageTextConverter, FirstLetterConverter -from pyrit.prompt_normalizer import PromptConverterConfiguration +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.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 @@ -24,6 +20,7 @@ from pathlib import Path from pyrit.scenario.core.atomic_attack import AtomicAttack + from pyrit.scenario.core.attack_technique_factory import AttackTechniqueFactory from pyrit.scenario.core.scenario_context import ScenarioContext from pyrit.scenario.core.scenario_strategy import ScenarioStrategy from pyrit.score import TrueFalseScorer @@ -34,32 +31,23 @@ # Leakage-specific technique catalog # --------------------------------------------------------------------------- -_BLANK_IMAGE_PATH = str(DATASETS_PATH / "seed_datasets" / "local" / "examples" / "blank_canvas.png") - -LEAKAGE_FACTORIES: list[AttackTechniqueFactory] = [ - AttackTechniqueFactory( - name="first_letter", - attack_class=PromptSendingAttack, - strategy_tags=["single_turn", "default"], - attack_kwargs={ - "attack_converter_config": AttackConverterConfig( - request_converters=PromptConverterConfiguration.from_converters(converters=[FirstLetterConverter()]) - ), - }, - ), - AttackTechniqueFactory( - name="image", - attack_class=PromptSendingAttack, - strategy_tags=["single_turn", "default"], - attack_kwargs={ - "attack_converter_config": AttackConverterConfig( - request_converters=PromptConverterConfiguration.from_converters( - converters=[AddImageTextConverter(img_to_add=_BLANK_IMAGE_PATH)] - ) - ), - }, - ), -] + +@cache +def _leakage_factories() -> list[AttackTechniqueFactory]: + """ + Return the AIRT source-owned leakage techniques (``first_letter``, ``image``). + + Imported lazily from the shared catalog (``techniques.airt``) to avoid an + import cycle during ``pyrit.scenario`` package initialization. These live in + the catalog but are not registered into the global registry — Leakage passes + them explicitly, so the shared technique pool for other scenarios is unchanged. + + Returns: + list[AttackTechniqueFactory]: The leakage-owned technique factories. + """ + from pyrit.setup.initializers.techniques.airt import get_technique_factories + + return get_technique_factories() @cache @@ -75,15 +63,15 @@ def _build_leakage_strategy() -> type[ScenarioStrategy]: """ registry = AttackTechniqueRegistry.get_registry_singleton() core_factories = list(registry.get_factories_or_raise().values()) - all_factories = core_factories + LEAKAGE_FACTORIES + all_factories = core_factories + _leakage_factories() return AttackTechniqueRegistry.build_strategy_class_from_factories( # type: ignore[return-value, ty:invalid-return-type] class_name="LeakageStrategy", factories=all_factories, aggregate_tags={ - "default": TagQuery.any_of("default"), "single_turn": TagQuery.any_of("single_turn"), "multi_turn": TagQuery.any_of("multi_turn"), }, + default_technique_names={"role_play", "many_shot", "first_letter", "image"}, ) @@ -161,5 +149,5 @@ async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list context=context, objective_scorer=self._objective_scorer, strategy_converters=self._strategy_converters, - extra_factories={factory.name: factory for factory in LEAKAGE_FACTORIES}, + extra_factories={factory.name: factory for factory in _leakage_factories()}, ) diff --git a/pyrit/scenario/scenarios/airt/rapid_response.py b/pyrit/scenario/scenarios/airt/rapid_response.py index 652a1e9711..a978620c62 100644 --- a/pyrit/scenario/scenarios/airt/rapid_response.py +++ b/pyrit/scenario/scenarios/airt/rapid_response.py @@ -49,12 +49,13 @@ def _build_rapid_response_strategy() -> type[ScenarioStrategy]: return AttackTechniqueRegistry.build_strategy_class_from_factories( # type: ignore[ty:invalid-return-type] class_name="RapidResponseStrategy", - factories=TagQuery.all("core").filter(factories), + factories=factories, + available=TagQuery.all("core"), aggregate_tags={ - "default": TagQuery.any_of("default"), "single_turn": TagQuery.any_of("single_turn"), "multi_turn": TagQuery.any_of("multi_turn"), }, + default_technique_names={"role_play", "many_shot"}, ) diff --git a/pyrit/scenario/scenarios/benchmark/adversarial.py b/pyrit/scenario/scenarios/benchmark/adversarial.py index 93480c9898..880161971c 100644 --- a/pyrit/scenario/scenarios/benchmark/adversarial.py +++ b/pyrit/scenario/scenarios/benchmark/adversarial.py @@ -62,11 +62,11 @@ def _build_benchmark_strategy() -> type[ScenarioStrategy]: class_name="BenchmarkStrategy", factories=factories, aggregate_tags={ - "default": TagQuery.any_of("default"), "light": TagQuery.any_of("light"), "single_turn": TagQuery.any_of("single_turn"), "multi_turn": TagQuery.any_of("multi_turn"), }, + default_technique_names={"role_play", "many_shot"}, ) diff --git a/pyrit/scenario/scenarios/garak/doctor.py b/pyrit/scenario/scenarios/garak/doctor.py index 629dd17ea0..7bd5166bca 100644 --- a/pyrit/scenario/scenarios/garak/doctor.py +++ b/pyrit/scenario/scenarios/garak/doctor.py @@ -10,55 +10,31 @@ from pyrit.executor.attack import AttackConverterConfig, PromptSendingAttack from pyrit.prompt_converter import LeetspeakConverter, PolicyPuppetryConverter, PolicyPuppetryTemplate from pyrit.prompt_normalizer import PromptConverterConfiguration +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 DatasetAttackConfiguration from pyrit.scenario.core.matrix_atomic_attack_builder import MatrixAtomicAttackBuilder from pyrit.scenario.core.scenario import BaselineAttackPolicy, Scenario -from pyrit.scenario.core.scenario_strategy import ScenarioStrategy if TYPE_CHECKING: 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 logger = logging.getLogger(__name__) -class DoctorStrategy(ScenarioStrategy): - """ - Strategies for the Doctor scenario. - - Each strategy applies a Policy Puppetry prompt-injection template to the - objective. ``PolicyPuppetry`` wraps the objective in the universal Dr House - TV-script template; ``PolicyPuppetryLeet`` additionally leetspeak-encodes the - templated prompt. - """ - - # Aggregate members - ALL = ("all", {"all"}) - DEFAULT = ("default", {"default"}) - - # Concrete strategies (values match the technique factory names). Both are tagged - # "default", so DEFAULT and ALL coincide today; ALL exists so a future non-default - # technique would diverge from DEFAULT without another default-strategy change. - PolicyPuppetry = ("policy_puppetry", {"default"}) - PolicyPuppetryLeet = ("policy_puppetry_leet", {"default"}) - - @classmethod - def get_aggregate_tags(cls) -> set[str]: - """Return the aggregate tags for the Doctor scenario.""" - return super().get_aggregate_tags() | {"default"} - - # Doctor-specific technique factories. Kept local to this scenario (referenced from -# _build_atomic_attacks_async) so they don't pollute the global registry. +# _build_atomic_attacks_async) so they don't pollute the global registry — the Policy +# Puppetry templates are pinned to this probe rather than being general-purpose. # The Dr House template is pinned (matching Garak's "Bypass" probe) so the # scenario stays deterministic rather than using the converter's random default. DOCTOR_FACTORIES: list[AttackTechniqueFactory] = [ AttackTechniqueFactory( name="policy_puppetry", attack_class=PromptSendingAttack, - strategy_tags=["default"], + strategy_tags=["single_turn"], attack_kwargs={ "attack_converter_config": AttackConverterConfig( request_converters=PromptConverterConfiguration.from_converters( @@ -72,7 +48,7 @@ def get_aggregate_tags(cls) -> set[str]: AttackTechniqueFactory( name="policy_puppetry_leet", attack_class=PromptSendingAttack, - strategy_tags=["default"], + strategy_tags=["single_turn"], attack_kwargs={ "attack_converter_config": AttackConverterConfig( request_converters=PromptConverterConfiguration.from_converters( @@ -87,6 +63,28 @@ def get_aggregate_tags(cls) -> set[str]: ] +# Doctor's strategy enum is generated from DOCTOR_FACTORIES via the shared factory +# generator (like the registry-driven scenarios) rather than hand-written. Both +# techniques are the scenario default, so DEFAULT and ALL coincide today; ALL exists +# so a future non-default technique would diverge from DEFAULT without another change. +def _build_doctor_strategy() -> type[ScenarioStrategy]: + """ + Generate the Doctor strategy enum from ``DOCTOR_FACTORIES``. + + Returns: + type[ScenarioStrategy]: The dynamically generated strategy enum class. + """ + return AttackTechniqueRegistry.build_strategy_class_from_factories( # type: ignore[return-value, ty:invalid-return-type] + class_name="DoctorStrategy", + factories=DOCTOR_FACTORIES, + aggregate_tags={}, + default_technique_names={"policy_puppetry", "policy_puppetry_leet"}, + ) + + +DoctorStrategy: type[ScenarioStrategy] = _build_doctor_strategy() + + class Doctor(Scenario): """ Doctor scenario implementation for PyRIT. @@ -136,7 +134,7 @@ def __init__( super().__init__( version=self.VERSION, strategy_class=DoctorStrategy, - default_strategy=DoctorStrategy.DEFAULT, + default_strategy=DoctorStrategy("default"), default_dataset_config=DatasetAttackConfiguration(dataset_names=["garak_doctor"]), objective_scorer=objective_scorer, scenario_result_id=scenario_result_id, diff --git a/pyrit/setup/initialization.py b/pyrit/setup/initialization.py index 43e7bfd4e0..eb0cf04ff8 100644 --- a/pyrit/setup/initialization.py +++ b/pyrit/setup/initialization.py @@ -200,6 +200,7 @@ async def initialize_pyrit_async( *, initialization_scripts: Sequence[str | pathlib.Path] | None = None, initializers: Sequence["PyRITInitializer"] | None = None, + load_defaults: bool = True, env_files: Sequence[pathlib.Path] | None = None, env_akv_ref: Sequence[str] | None = None, silent: bool = False, @@ -216,6 +217,15 @@ async def initialize_pyrit_async( loaded and executed. Loading is handled by the InitializerRegistry. initializers (Sequence[PyRITInitializer] | None): Optional sequence of PyRITInitializer instances to execute directly. These provide type-safe, validated configuration with clear documentation. + load_defaults (bool): If True (default) AND the caller supplies neither ``initializers`` nor + ``initialization_scripts``, a default initializer set is run so a bare + ``initialize_pyrit_async(...)`` yields a usable environment: the core attack-technique catalog + (``TechniqueInitializer``, populating the AttackTechniqueRegistry) plus the available default + targets (``TargetInitializer``, registering whatever endpoints are configured via env vars). + Supplying any initializer or script means the caller owns setup, so the defaults are skipped; + set this to False to also skip them on a bare call (e.g. to start from an empty state). Only the + ``core`` techniques and ``default`` targets are loaded — ``extra`` / per-source technique groups + and ``scorer`` target variants remain opt-in. env_files (Sequence[pathlib.Path] | None): Optional sequence of environment file paths to load in order. If not provided, will load default .env and .env.local files from PyRIT home if they exist. All paths must be valid pathlib.Path objects. @@ -259,8 +269,8 @@ async def initialize_pyrit_async( CentralMemory.set_memory_instance(memory) - # Combine directly provided initializers with those loaded from scripts - all_initializers = list(initializers) if initializers else [] + # Combine directly provided initializers with those loaded from scripts. + all_initializers: list[PyRITInitializer] = list(initializers) if initializers else [] # Load additional initializers from scripts — the registry owns turning # external script files into initializer instances. @@ -271,6 +281,16 @@ async def initialize_pyrit_async( script_initializers = registry.create_from_script_paths(script_paths=initialization_scripts) all_initializers.extend(script_initializers) + # When the caller supplies nothing, fall back to the default initializer set so a + # bare initialize_pyrit_async(...) yields a usable environment (core techniques + + # available default targets). Supplying any initializer/script means the caller owns + # setup, so defaults are skipped; load_defaults=False skips them even on a bare call. + if load_defaults and not all_initializers: + from pyrit.setup.initializers.targets import TargetInitializer + from pyrit.setup.initializers.techniques import TechniqueInitializer + + all_initializers = [TechniqueInitializer(), TargetInitializer()] + # Execute all initializers in order if all_initializers: await _execute_initializers_async(initializers=all_initializers) diff --git a/pyrit/setup/initializers/techniques/airt.py b/pyrit/setup/initializers/techniques/airt.py new file mode 100644 index 0000000000..45d23f90eb --- /dev/null +++ b/pyrit/setup/initializers/techniques/airt.py @@ -0,0 +1,62 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +""" +AIRT source-owned scenario techniques. + +Home for techniques that belong to an AIRT scenario/source but are not part of +the general-purpose ``core`` catalog. They live in the shared catalog (tagged +with the ``airt`` owner) so any scenario can select them, yet default to their +owning scenario. + +Unlike ``core``/``extra`` (registered into the global registry by +``TechniqueInitializer``), these are imported directly by their owning scenario +and are intentionally *not* part of the default ``build_technique_factories`` +aggregation — so scenarios that consume the full catalog keep their existing +technique pool. +""" + +from pyrit.common.path import DATASETS_PATH +from pyrit.executor.attack import AttackConverterConfig, PromptSendingAttack +from pyrit.prompt_converter import AddImageTextConverter, FirstLetterConverter +from pyrit.prompt_normalizer import PromptConverterConfiguration +from pyrit.scenario.core.attack_technique_factory import AttackTechniqueFactory + +_BLANK_IMAGE_PATH = str(DATASETS_PATH / "seed_datasets" / "local" / "examples" / "blank_canvas.png") + + +def get_technique_factories() -> list[AttackTechniqueFactory]: + """ + Build the AIRT source-owned technique factories. + + These back the Leakage scenario (``first_letter`` obfuscation and a blank-image + carrier). They carry a scenario-specific ``leakage`` tag (they're quite specific + to leakage) while staying selectable by any scenario via their ``airt`` owner tag. + + Returns: + list[AttackTechniqueFactory]: The AIRT source-owned techniques. + """ + return [ + AttackTechniqueFactory( + name="first_letter", + attack_class=PromptSendingAttack, + strategy_tags=["single_turn", "airt", "leakage"], + attack_kwargs={ + "attack_converter_config": AttackConverterConfig( + request_converters=PromptConverterConfiguration.from_converters(converters=[FirstLetterConverter()]) + ), + }, + ), + AttackTechniqueFactory( + name="image", + attack_class=PromptSendingAttack, + strategy_tags=["single_turn", "airt", "leakage"], + attack_kwargs={ + "attack_converter_config": AttackConverterConfig( + request_converters=PromptConverterConfiguration.from_converters( + converters=[AddImageTextConverter(img_to_add=_BLANK_IMAGE_PATH)] + ) + ), + }, + ), + ] diff --git a/pyrit/setup/initializers/techniques/core.py b/pyrit/setup/initializers/techniques/core.py index bc93592869..dd2394ce1c 100644 --- a/pyrit/setup/initializers/techniques/core.py +++ b/pyrit/setup/initializers/techniques/core.py @@ -4,10 +4,15 @@ """ Core scenario techniques. -Exposes ``get_technique_factories()`` returning the default catalog of -attack technique factories. The ``core`` group tag is injected by -``build_technique_factories`` — factories here carry only their behavioral -tags (e.g. ``single_turn``/``multi_turn``/``default``/``light``). +``core`` is the home for general-purpose attack techniques usable by any +scenario. The ``core`` group tag is injected by ``build_technique_factories`` — +factories here carry only their behavioral tags (e.g. +``single_turn``/``multi_turn``/``light``). + +``default`` is intentionally not a tag here: what runs by default is +scenario-relative and is declared per scenario (see +``AttackTechniqueRegistry.build_strategy_class_from_factories``'s +``default_technique_names``), not baked into the shared catalog. """ from pyrit.executor.attack import ( @@ -25,15 +30,15 @@ def get_technique_factories() -> list[AttackTechniqueFactory]: """ Build the core scenario technique factories. - Factories that need an adversarial chat target do not bake one in; the - default adversarial target is resolved lazily inside - ``AttackTechniqueFactory.create`` via ``get_default_adversarial_target()``. - A bare ``PromptSendingAttack`` factory is intentionally omitted: every scenario whose ``BASELINE_ATTACK_POLICY`` is ``BaselineAttackPolicy.Enabled`` already auto-prepends an equivalent baseline atomic attack via ``Scenario._build_baseline_atomic_attack``. + Factories that need an adversarial chat target do not bake one in; the + default adversarial target is resolved lazily inside + ``AttackTechniqueFactory.create`` via ``get_default_adversarial_target()``. + Returns: list[AttackTechniqueFactory]: The core scenario techniques. """ @@ -41,13 +46,13 @@ def get_technique_factories() -> list[AttackTechniqueFactory]: AttackTechniqueFactory( name="role_play", attack_class=RolePlayAttack, - strategy_tags=["single_turn", "default", "light"], + strategy_tags=["single_turn", "light"], attack_kwargs={"role_play_definition_path": RolePlayPaths.MOVIE_SCRIPT.value}, ), AttackTechniqueFactory( name="many_shot", attack_class=ManyShotJailbreakAttack, - strategy_tags=["multi_turn", "default", "light"], + strategy_tags=["multi_turn", "light"], ), AttackTechniqueFactory( name="tap", @@ -58,16 +63,6 @@ def get_technique_factories() -> list[AttackTechniqueFactory]: name="crescendo_simulated", strategy_tags=["single_turn"], ), - AttackTechniqueFactory( - name="red_teaming", - attack_class=RedTeamingAttack, - strategy_tags=["multi_turn", "light"], - ), - AttackTechniqueFactory( - name="context_compliance", - attack_class=ContextComplianceAttack, - strategy_tags=["single_turn", "light"], - ), AttackTechniqueFactory.with_simulated_conversation( name="crescendo_movie_director", strategy_tags=["single_turn"], @@ -80,4 +75,14 @@ def get_technique_factories() -> list[AttackTechniqueFactory]: name="crescendo_journalist_interview", strategy_tags=["single_turn"], ), + AttackTechniqueFactory( + name="red_teaming", + attack_class=RedTeamingAttack, + strategy_tags=["multi_turn", "light"], + ), + AttackTechniqueFactory( + name="context_compliance", + attack_class=ContextComplianceAttack, + strategy_tags=["single_turn", "light"], + ), ] diff --git a/tests/unit/scenario/garak/test_doctor.py b/tests/unit/scenario/garak/test_doctor.py index ebaad7601c..2e62ea8bf6 100644 --- a/tests/unit/scenario/garak/test_doctor.py +++ b/tests/unit/scenario/garak/test_doctor.py @@ -191,5 +191,5 @@ def test_get_aggregate_tags_includes_default(self): assert "default" in DoctorStrategy.get_aggregate_tags() def test_concrete_strategy_values(self): - assert DoctorStrategy.PolicyPuppetry.value == "policy_puppetry" - assert DoctorStrategy.PolicyPuppetryLeet.value == "policy_puppetry_leet" + assert DoctorStrategy("policy_puppetry").value == "policy_puppetry" + assert DoctorStrategy("policy_puppetry_leet").value == "policy_puppetry_leet" From 5cfd1a9332b91d5acb57698d6724f22705febd34 Mon Sep 17 00:00:00 2001 From: Richard Lundeen Date: Thu, 9 Jul 2026 16:36:27 -0700 Subject: [PATCH 2/2] MAINT: build DoctorStrategy in __init__ for consistency Address review feedback: build the Doctor strategy class the same way the other dynamically-generated scenarios do (rapid_response/leakage/cyber) instead of at module level. - `_build_doctor_strategy()` is now `@cache`d and called in `Doctor.__init__` (`strategy_class = _build_doctor_strategy()`; `default_strategy=strategy_class("default")`). - The public `DoctorStrategy` symbol is resolved lazily via a `__getattr__` in the garak package `__init__` (mirroring the airt package), so `@cache` keeps every instance sharing one enum class. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- pyrit/scenario/scenarios/garak/__init__.py | 20 +++++++++++++++++++- pyrit/scenario/scenarios/garak/doctor.py | 14 +++++++++----- 2 files changed, 28 insertions(+), 6 deletions(-) diff --git a/pyrit/scenario/scenarios/garak/__init__.py b/pyrit/scenario/scenarios/garak/__init__.py index b394c8e5bf..963b68b445 100644 --- a/pyrit/scenario/scenarios/garak/__init__.py +++ b/pyrit/scenario/scenarios/garak/__init__.py @@ -3,10 +3,28 @@ """Garak-based attack scenarios.""" -from pyrit.scenario.scenarios.garak.doctor import Doctor, DoctorStrategy +from typing import Any + +from pyrit.scenario.scenarios.garak.doctor import Doctor, _build_doctor_strategy from pyrit.scenario.scenarios.garak.encoding import Encoding, EncodingStrategy from pyrit.scenario.scenarios.garak.web_injection import WebInjection, WebInjectionStrategy + +def __getattr__(name: str) -> Any: + """ + Lazily resolve the dynamically-generated Doctor strategy class. + + Returns: + Any: The resolved strategy class. + + Raises: + AttributeError: If the attribute name is not recognized. + """ + if name == "DoctorStrategy": + return _build_doctor_strategy() + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") + + __all__ = [ "Doctor", "DoctorStrategy", diff --git a/pyrit/scenario/scenarios/garak/doctor.py b/pyrit/scenario/scenarios/garak/doctor.py index 7bd5166bca..a41b9c5fd3 100644 --- a/pyrit/scenario/scenarios/garak/doctor.py +++ b/pyrit/scenario/scenarios/garak/doctor.py @@ -4,6 +4,7 @@ from __future__ import annotations import logging +from functools import cache from typing import TYPE_CHECKING, ClassVar from pyrit.common import apply_defaults @@ -67,6 +68,10 @@ # generator (like the registry-driven scenarios) rather than hand-written. Both # techniques are the scenario default, so DEFAULT and ALL coincide today; ALL exists # so a future non-default technique would diverge from DEFAULT without another change. +# Built lazily and cached (like the other dynamically-generated scenarios) so every +# Doctor instance shares one enum class; the public ``DoctorStrategy`` symbol is +# resolved from here via the garak package ``__getattr__``. +@cache def _build_doctor_strategy() -> type[ScenarioStrategy]: """ Generate the Doctor strategy enum from ``DOCTOR_FACTORIES``. @@ -82,9 +87,6 @@ def _build_doctor_strategy() -> type[ScenarioStrategy]: ) -DoctorStrategy: type[ScenarioStrategy] = _build_doctor_strategy() - - class Doctor(Scenario): """ Doctor scenario implementation for PyRIT. @@ -131,10 +133,12 @@ def __init__( if not objective_scorer: objective_scorer = self._get_default_objective_scorer() + strategy_class = _build_doctor_strategy() + super().__init__( version=self.VERSION, - strategy_class=DoctorStrategy, - default_strategy=DoctorStrategy("default"), + strategy_class=strategy_class, + default_strategy=strategy_class("default"), default_dataset_config=DatasetAttackConfiguration(dataset_names=["garak_doctor"]), objective_scorer=objective_scorer, scenario_result_id=scenario_result_id,