Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 7 additions & 3 deletions doc/code/framework.md
Original file line number Diff line number Diff line change
Expand Up @@ -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)

Expand Down
67 changes: 57 additions & 10 deletions pyrit/registry/components/attack_technique_registry.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
2 changes: 1 addition & 1 deletion pyrit/scenario/scenarios/adaptive/text_adaptive.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"},
)


Expand Down
20 changes: 10 additions & 10 deletions pyrit/scenario/scenarios/airt/cyber.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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,
)


Expand Down
56 changes: 22 additions & 34 deletions pyrit/scenario/scenarios/airt/leakage.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand All @@ -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
Expand All @@ -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"},
)


Expand Down Expand Up @@ -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()},
)
5 changes: 3 additions & 2 deletions pyrit/scenario/scenarios/airt/rapid_response.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"},
)


Expand Down
2 changes: 1 addition & 1 deletion pyrit/scenario/scenarios/benchmark/adversarial.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"},
)


Expand Down
20 changes: 19 additions & 1 deletion pyrit/scenario/scenarios/garak/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down
Loading
Loading