diff --git a/doc/code/framework.md b/doc/code/framework.md index 3f646e4138..c150eaad04 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 `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). - **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 3d5042f228..ffa0bbbca0 100644 --- a/pyrit/registry/components/attack_technique_registry.py +++ b/pyrit/registry/components/attack_technique_registry.py @@ -178,44 +178,91 @@ 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, ) -> 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 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 ``ScenarioTechnique`` subclass with the generated members. """ 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) + + # 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))} + 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.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) # Build the enum class dynamically diff --git a/pyrit/scenario/scenarios/adaptive/text_adaptive.py b/pyrit/scenario/scenarios/adaptive/text_adaptive.py index 86c3f8314c..6e1b9d0069 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_technique() -> type[ScenarioTechnique]: class_name="TextAdaptiveTechnique", 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 a5aca83456..51debd1df9 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_technique), 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_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_DEFAULT_TECHNIQUE_NAMES = {"red_teaming"} @cache @@ -58,13 +59,12 @@ def _build_cyber_technique() -> type[ScenarioTechnique]: class_name="CyberTechnique", 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 26608363fa..504fc2bf3d 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_technique import ScenarioTechnique 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, - technique_tags=["single_turn", "default"], - attack_kwargs={ - "attack_converter_config": AttackConverterConfig( - request_converters=PromptConverterConfiguration.from_converters(converters=[FirstLetterConverter()]) - ), - }, - ), - AttackTechniqueFactory( - name="image", - attack_class=PromptSendingAttack, - technique_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_technique() -> type[ScenarioTechnique]: """ 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_technique_class_from_factories( # type: ignore[return-value, ty:invalid-return-type] class_name="LeakageTechnique", 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, technique_converters=self._technique_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 fe3712be8d..4a4635a271 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_technique() -> type[ScenarioTechnique]: return AttackTechniqueRegistry.build_technique_class_from_factories( # type: ignore[ty:invalid-return-type] class_name="RapidResponseTechnique", - 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 40ab4c579f..ec74616fc5 100644 --- a/pyrit/scenario/scenarios/benchmark/adversarial.py +++ b/pyrit/scenario/scenarios/benchmark/adversarial.py @@ -62,11 +62,11 @@ def _build_benchmark_technique() -> type[ScenarioTechnique]: class_name="BenchmarkTechnique", 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/__init__.py b/pyrit/scenario/scenarios/garak/__init__.py index 21e31ad7c0..9c2575dc67 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, DoctorTechnique +from typing import Any + +from pyrit.scenario.scenarios.garak.doctor import Doctor, _build_doctor_technique from pyrit.scenario.scenarios.garak.encoding import Encoding, EncodingTechnique from pyrit.scenario.scenarios.garak.web_injection import WebInjection, WebInjectionTechnique + +def __getattr__(name: str) -> Any: + """ + Lazily resolve the dynamically-generated Doctor technique class. + + Returns: + Any: The resolved technique class. + + Raises: + AttributeError: If the attribute name is not recognized. + """ + if name == "DoctorTechnique": + return _build_doctor_technique() + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") + + __all__ = [ "Doctor", "DoctorTechnique", diff --git a/pyrit/scenario/scenarios/garak/doctor.py b/pyrit/scenario/scenarios/garak/doctor.py index 066d445623..ed7b84ca01 100644 --- a/pyrit/scenario/scenarios/garak/doctor.py +++ b/pyrit/scenario/scenarios/garak/doctor.py @@ -4,61 +4,38 @@ from __future__ import annotations import logging +from functools import cache from typing import TYPE_CHECKING, ClassVar from pyrit.common import apply_defaults 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_technique import ScenarioTechnique if TYPE_CHECKING: from pyrit.scenario.core.atomic_attack import AtomicAttack from pyrit.scenario.core.scenario_context import ScenarioContext + from pyrit.scenario.core.scenario_technique import ScenarioTechnique from pyrit.score import TrueFalseScorer logger = logging.getLogger(__name__) -class DoctorTechnique(ScenarioTechnique): - """ - Techniques for the Doctor scenario. - - Each technique 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 techniques (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-technique 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, - technique_tags=["default"], + technique_tags=["single_turn"], attack_kwargs={ "attack_converter_config": AttackConverterConfig( request_converters=PromptConverterConfiguration.from_converters( @@ -72,7 +49,7 @@ def get_aggregate_tags(cls) -> set[str]: AttackTechniqueFactory( name="policy_puppetry_leet", attack_class=PromptSendingAttack, - technique_tags=["default"], + technique_tags=["single_turn"], attack_kwargs={ "attack_converter_config": AttackConverterConfig( request_converters=PromptConverterConfiguration.from_converters( @@ -87,6 +64,29 @@ def get_aggregate_tags(cls) -> set[str]: ] +# Doctor's technique 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. +# Built lazily and cached (like the other dynamically-generated scenarios) so every +# Doctor instance shares one enum class; the public ``DoctorTechnique`` symbol is +# resolved from here via the garak package ``__getattr__``. +@cache +def _build_doctor_technique() -> type[ScenarioTechnique]: + """ + Generate the Doctor technique enum from ``DOCTOR_FACTORIES``. + + Returns: + type[ScenarioTechnique]: The dynamically generated technique enum class. + """ + 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"}, + ) + + class Doctor(Scenario): """ Doctor scenario implementation for PyRIT. @@ -133,10 +133,12 @@ def __init__( if not objective_scorer: objective_scorer = self._get_default_objective_scorer() + technique_class = _build_doctor_technique() + super().__init__( version=self.VERSION, - technique_class=DoctorTechnique, - default_technique=DoctorTechnique.DEFAULT, + 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/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..73ce450232 --- /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, + technique_tags=["single_turn", "airt", "leakage"], + attack_kwargs={ + "attack_converter_config": AttackConverterConfig( + request_converters=PromptConverterConfiguration.from_converters(converters=[FirstLetterConverter()]) + ), + }, + ), + AttackTechniqueFactory( + name="image", + attack_class=PromptSendingAttack, + technique_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 aaf07a8b6f..4507697454 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_technique_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, - technique_tags=["single_turn", "default", "light"], + technique_tags=["single_turn", "light"], attack_kwargs={"role_play_definition_path": RolePlayPaths.MOVIE_SCRIPT.value}, ), AttackTechniqueFactory( name="many_shot", attack_class=ManyShotJailbreakAttack, - technique_tags=["multi_turn", "default", "light"], + technique_tags=["multi_turn", "light"], ), AttackTechniqueFactory( name="tap", @@ -58,16 +63,6 @@ def get_technique_factories() -> list[AttackTechniqueFactory]: name="crescendo_simulated", technique_tags=["single_turn"], ), - AttackTechniqueFactory( - name="red_teaming", - attack_class=RedTeamingAttack, - technique_tags=["multi_turn", "light"], - ), - AttackTechniqueFactory( - name="context_compliance", - attack_class=ContextComplianceAttack, - technique_tags=["single_turn", "light"], - ), AttackTechniqueFactory.with_simulated_conversation( name="crescendo_movie_director", technique_tags=["single_turn"], @@ -80,4 +75,14 @@ def get_technique_factories() -> list[AttackTechniqueFactory]: name="crescendo_journalist_interview", technique_tags=["single_turn"], ), + AttackTechniqueFactory( + name="red_teaming", + attack_class=RedTeamingAttack, + technique_tags=["multi_turn", "light"], + ), + AttackTechniqueFactory( + name="context_compliance", + attack_class=ContextComplianceAttack, + technique_tags=["single_turn", "light"], + ), ] diff --git a/tests/end_to_end/test_config.yaml b/tests/end_to_end/test_config.yaml index 2007c96299..2c2acc8e99 100644 --- a/tests/end_to_end/test_config.yaml +++ b/tests/end_to_end/test_config.yaml @@ -4,8 +4,8 @@ memory_db_type: in_memory # The catalog endpoint (GET /api/scenarios/catalog[/]) instantiates every # scenario class to build metadata. Most scenarios rely on the -# AttackTechniqueRegistry (e.g. airt.cyber via _build_cyber_strategy, airt.leakage -# via _build_leakage_strategy) and will raise at __init__ time if no factories have +# AttackTechniqueRegistry (e.g. airt.cyber via _build_cyber_technique, airt.leakage +# via _build_leakage_technique) and will raise at __init__ time if no factories have # been registered yet, so the canonical attack technique factories must be loaded # before the first catalog request. Per-scenario run initializers # (target, load_default_datasets) are still passed via the CLI invocation in diff --git a/tests/unit/scenario/garak/test_doctor.py b/tests/unit/scenario/garak/test_doctor.py index b7b01fa517..25004cbaec 100644 --- a/tests/unit/scenario/garak/test_doctor.py +++ b/tests/unit/scenario/garak/test_doctor.py @@ -200,5 +200,5 @@ def test_get_aggregate_tags_includes_default(self): assert "default" in DoctorTechnique.get_aggregate_tags() def test_concrete_technique_values(self): - assert DoctorTechnique.PolicyPuppetry.value == "policy_puppetry" - assert DoctorTechnique.PolicyPuppetryLeet.value == "policy_puppetry_leet" + assert DoctorTechnique("policy_puppetry").value == "policy_puppetry" + assert DoctorTechnique("policy_puppetry_leet").value == "policy_puppetry_leet"