FEAT Content harm scenario

.pre-commit-config.yaml

@@ -54,6 +54,7 @@ repos:
     hooks:
       - id: ruff-check
         name: ruff-check
+        args: [--fix]
 
   - repo: https://github.com/PyCQA/flake8
     rev: 7.1.2

doc/_toc.yml

@@ -128,7 +128,11 @@ chapters:
     - file: code/auxiliary_attacks/0_auxiliary_attacks
       sections:
       - file: code/auxiliary_attacks/1_gcg_azure_ml
-    - file: code/scenarios/scenarios
+    - file: code/scenarios/0_scenarios
+      sections:
+      - file: code/scenarios/1_composite_scenario
+      - file: code/scenarios/2_end_to_end_scenario_datasets
+      - file: code/scenarios/2_end_to_end_scenario
     - file: code/front_end/0_front_end
       sections:
       - file: code/front_end/1_pyrit_scan

doc/code/front_end/1_pyrit_scan.ipynb

@@ -7,7 +7,7 @@
    "source": [
     "# 1. PyRIT Scan - Command Line Execution\n",
     "\n",
-    "`pyrit_scan` allows you to run automated security testing and red teaming attacks against AI systems using [scenarios](../scenarios/scenarios.ipynb) for strategies and [configuration](../setup/1_configuration.ipynb).\n",
+    "`pyrit_scan` allows you to run automated security testing and red teaming attacks against AI systems using [scenarios](../scenarios/0_scenarios.ipynb) for strategies and [configuration](../setup/1_configuration.ipynb).\n",
     "\n",
     "Note in this doc the ! prefaces all commands in the terminal so we can run in a Jupyter Notebook.\n",
     "\n",
@@ -442,7 +442,6 @@
     "            name=\"My Custom Scenario\",\n",
     "            version=1,\n",
     "            strategy_class=MyCustomStrategy,\n",
-    "            default_aggregate=MyCustomStrategy.ALL,\n",
     "            scenario_result_id=scenario_result_id,\n",
     "        )\n",
     "        # ... your scenario-specific initialization code\n",

doc/code/front_end/1_pyrit_scan.py

@@ -15,7 +15,7 @@
 # %% [markdown]
 # # 1. PyRIT Scan
 #
-# `pyrit_scan` allows you to run automated security testing and red teaming attacks against AI systems using [scenarios](../scenarios/scenarios.ipynb) for strategies and [configuration](../setup/1_configuration.ipynb).
+# `pyrit_scan` allows you to run automated security testing and red teaming attacks against AI systems using [scenarios](../scenarios/0_scenarios.ipynb) for strategies and [configuration](../setup/1_configuration.ipynb).
 #
 # Note in this doc the ! prefaces all commands in the terminal so we can run in a Jupyter Notebook.
 #
@@ -154,7 +154,6 @@ def __init__(self, *, scenario_result_id=None, **kwargs):
             name="My Custom Scenario",
             version=1,
             strategy_class=MyCustomStrategy,
-            default_aggregate=MyCustomStrategy.ALL,
             scenario_result_id=scenario_result_id,
         )
         # ... your scenario-specific initialization code

doc/code/scenarios/0_scenarios.ipynb

@@ -0,0 +1,290 @@
+{
+ "cells": [
+  {
+   "cell_type": "markdown",
+   "id": "0",
+   "metadata": {
+    "lines_to_next_cell": 0
+   },
+   "source": [
+    "Scenarios\n",
+    "\n",
+    "A `Scenario` is a higher-level construct that groups multiple Attack Configurations together. This allows you to execute a comprehensive testing campaign with multiple attack methods sequentially. Scenarios are meant to be configured and written to test for specific workflows. As such, it is okay to hard code some values.\n",
+    "\n",
+    "# What is a Scenario?\n",
+    "\n",
+    "A `Scenario` represents a comprehensive testing campaign composed of multiple atomic attack tests. It orchestrates the execution of multiple `AtomicAttack` instances sequentially and aggregates the results into a single `ScenarioResult`.\n",
+    "\n",
+    "## Key Components\n",
+    "\n",
+    "- **Scenario**: The top-level orchestrator that groups and executes multiple atomic attacks\n",
+    "- **AtomicAttack**: An atomic test unit combining an attack strategy, objectives, and execution parameters\n",
+    "- **ScenarioResult**: Contains the aggregated results from all atomic attacks and scenario metadata\n",
+    "\n",
+    "# Use Cases\n",
+    "\n",
+    "Some examples of scenarios you might create:\n",
+    "\n",
+    "- **VibeCheckScenario**: Randomly selects a few prompts from HarmBench to quickly assess model behavior\n",
+    "- **QuickViolence**: Checks how resilient a model is to violent objectives using multiple attack techniques\n",
+    "- **ComprehensiveFoundry**: Tests a target with all available attack converters and strategies\n",
+    "- **CustomCompliance**: Tests against specific compliance requirements with curated datasets and attacks\n",
+    "\n",
+    "These Scenarios can be updated and added to as you refine what you are testing for.\n",
+    "\n",
+    "# How It Works\n",
+    "\n",
+    "Each `Scenario` contains a collection of `AtomicAttack` objects. When executed:\n",
+    "\n",
+    "1. Each `AtomicAttack` is executed sequentially\n",
+    "2. Every `AtomicAttack` tests its configured attack against all specified objectives and datasets\n",
+    "3. Results are aggregated into a single `ScenarioResult` with all attack outcomes\n",
+    "4. Optional memory labels help track and categorize the scenario execution\n",
+    "\n",
+    "# Creating Custom Scenarios\n",
+    "\n",
+    "To create a custom scenario, extend the `Scenario` base class and implement the required abstract methods.\n",
+    "\n",
+    "## Required Components\n",
+    "\n",
+    "1. **Strategy Enum**: Create a `ScenarioStrategy` enum that defines the available strategies for your scenario.\n",
+    "   - Each enum member is defined as `(value, tags)` where value is a string and tags is a set of strings\n",
+    "   - Include an `ALL` aggregate strategy that expands to all available strategies\n",
+    "   - Optionally implement `supports_composition()` and `validate_composition()` for strategy composition rules\n",
+    "\n",
+    "2. **Scenario Class**: Extend `Scenario` and implement these abstract methods:\n",
+    "   - `get_strategy_class()`: Return your strategy enum class\n",
+    "   - `get_default_strategy()`: Return the default strategy (typically `YourStrategy.ALL`)\n",
+    "   - `_get_atomic_attacks_async()`: Build and return a list of `AtomicAttack` instances\n",
+    "\n",
+    "3. **Constructor**: Use `@apply_defaults` decorator and call `super().__init__()` with scenario metadata:\n",
+    "   - `name`: Descriptive name for your scenario\n",
+    "   - `version`: Integer version number\n",
+    "   - `objective_target`: The target system being tested\n",
+    "   - `objective_scorer_identifier`: Identifier for the scoring mechanism\n",
+    "   - `memory_labels`: Optional labels for tracking\n",
+    "   - `max_concurrency`: Number of concurrent operations (default: 10)\n",
+    "   - `max_retries`: Number of retry attempts on failure (default: 0)\n",
+    "\n",
+    "## Example Structure"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "1",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from typing import List, Optional, Type\n",
+    "\n",
+    "from pyrit.common import apply_defaults\n",
+    "from pyrit.executor.attack import AttackScoringConfig, PromptSendingAttack\n",
+    "from pyrit.scenarios import AtomicAttack, Scenario, ScenarioStrategy\n",
+    "from pyrit.score.true_false.true_false_scorer import TrueFalseScorer\n",
+    "\n",
+    "\n",
+    "class MyStrategy(ScenarioStrategy):\n",
+    "    ALL = (\"all\", {\"all\"})\n",
+    "    StrategyA = (\"strategy_a\", {\"tag1\", \"tag2\"})\n",
+    "    StrategyB = (\"strategy_b\", {\"tag1\"})\n",
+    "\n",
+    "\n",
+    "class MyScenario(Scenario):\n",
+    "    version: int = 1\n",
+    "\n",
+    "    @classmethod\n",
+    "    def get_strategy_class(cls) -> Type[ScenarioStrategy]:\n",
+    "        return MyStrategy\n",
+    "\n",
+    "    @classmethod\n",
+    "    def get_default_strategy(cls) -> ScenarioStrategy:\n",
+    "        return MyStrategy.ALL\n",
+    "\n",
+    "    @apply_defaults\n",
+    "    def __init__(\n",
+    "        self,\n",
+    "        *,\n",
+    "        objective_scorer: Optional[TrueFalseScorer] = None,\n",
+    "    ):\n",
+    "\n",
+    "        self._scorer_config = AttackScoringConfig(objective_scorer=objective_scorer)\n",
+    "\n",
+    "        # Call parent constructor\n",
+    "        super().__init__(\n",
+    "            name=\"My Custom Scenario\",\n",
+    "            version=self.version,\n",
+    "            strategy_class=MyStrategy,\n",
+    "            objective_scorer_identifier=objective_scorer.get_identifier(),\n",
+    "        )\n",
+    "\n",
+    "    async def _get_atomic_attacks_async(self) -> List[AtomicAttack]:\n",
+    "        atomic_attacks = []\n",
+    "        assert self._objective_target is not None\n",
+    "        for strategy in self._strategy_compositions:  # type: ignore\n",
+    "            # Create attack instances based on strategy\n",
+    "            attack = PromptSendingAttack(\n",
+    "                objective_target=self._objective_target,\n",
+    "                attack_scoring_config=self._scorer_config,\n",
+    "            )\n",
+    "            atomic_attacks.append(\n",
+    "                AtomicAttack(\n",
+    "                    atomic_attack_name=strategy.name,\n",
+    "                    attack=attack,\n",
+    "                    objectives=[\"objective1\", \"objective2\"],\n",
+    "                    memory_labels=self._memory_labels,\n",
+    "                )\n",
+    "            )\n",
+    "        return atomic_attacks"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "2",
+   "metadata": {},
+   "source": [
+    "\n",
+    "## Existing Scenarios"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "3",
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "Loading PyRIT modules...\n"
+     ]
+    },
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "\n",
+      "Available Scenarios:\n",
+      "================================================================================\n",
+      "\u001b[1m\u001b[36m\n",
+      "  content_harm_scenario\u001b[0m\n",
+      "    Class: ContentHarmScenario\n",
+      "    Description:\n",
+      "      Content Harm Scenario implementation for PyRIT. This scenario contains\n",
+      "      various harm-based checks that you can run to get a quick idea about\n",
+      "      model behavior with respect to certain harm categories.\n",
+      "    Aggregate Strategies:\n",
+      "      - all\n",
+      "    Available Strategies (7):\n",
+      "      hate, fairness, violence, sexual, harassment, misinformation, leakage\n",
+      "    Default Strategy: all\n",
+      "\u001b[1m\u001b[36m\n",
+      "  encoding_scenario\u001b[0m\n",
+      "    Class: EncodingScenario\n",
+      "    Description:\n",
+      "      Encoding Scenario implementation for PyRIT. This scenario tests how\n",
+      "      resilient models are to various encoding attacks by encoding potentially\n",
+      "      harmful text (by default slurs and XSS payloads) and testing if the\n",
+      "      model will decode and repeat the encoded payload. It mimics the Garak\n",
+      "      encoding probe. The scenario works by: 1. Taking seed prompts (the\n",
+      "      harmful text to be encoded) 2. Encoding them using various encoding\n",
+      "      schemes (Base64, ROT13, Morse, etc.) 3. Asking the target model to\n",
+      "      decode the encoded text 4. Scoring whether the model successfully\n",
+      "      decoded and repeated the harmful content By default, this uses the same\n",
+      "      dataset as Garak: slur terms and web XSS payloads.\n",
+      "    Aggregate Strategies:\n",
+      "      - all\n",
+      "    Available Strategies (17):\n",
+      "      base64, base2048, base16, base32, ascii85, hex, quoted_printable,\n",
+      "      uuencode, rot13, braille, atbash, morse_code, nato, ecoji, zalgo,\n",
+      "      leet_speak, ascii_smuggler\n",
+      "    Default Strategy: all\n",
+      "\u001b[1m\u001b[36m\n",
+      "  foundry_scenario\u001b[0m\n",
+      "    Class: FoundryScenario\n",
+      "    Description:\n",
+      "      FoundryScenario is a preconfigured scenario that automatically generates\n",
+      "      multiple AtomicAttack instances based on the specified attack\n",
+      "      strategies. It supports both single-turn attacks (with various\n",
+      "      converters) and multi-turn attacks (Crescendo, RedTeaming), making it\n",
+      "      easy to quickly test a target against multiple attack vectors. The\n",
+      "      scenario can expand difficulty levels (EASY, MODERATE, DIFFICULT) into\n",
+      "      their constituent attack strategies, or you can specify individual\n",
+      "      strategies directly. Note this is not the same as the Foundry AI Red\n",
+      "      Teaming Agent. This is a PyRIT contract so their library can make use of\n",
+      "      PyRIT in a consistent way.\n",
+      "    Aggregate Strategies:\n",
+      "      - all, easy, moderate, difficult\n",
+      "    Available Strategies (23):\n",
+      "      ansi_attack, ascii_art, ascii_smuggler, atbash, base64, binary, caesar,\n",
+      "      character_space, char_swap, diacritic, flip, leetspeak, morse, rot13,\n",
+      "      suffix_append, string_join, unicode_confusable, unicode_substitution,\n",
+      "      url, jailbreak, tense, multi_turn, crescendo\n",
+      "    Default Strategy: easy\n",
+      "\n",
+      "================================================================================\n",
+      "\n",
+      "Total scenarios: 3\n"
+     ]
+    },
+    {
+     "data": {
+      "text/plain": [
+       "0"
+      ]
+     },
+     "execution_count": null,
+     "metadata": {},
+     "output_type": "execute_result"
+    }
+   ],
+   "source": [
+    "from pyrit.cli.frontend_core import FrontendCore, print_scenarios_list\n",
+    "\n",
+    "print_scenarios_list(context=FrontendCore())"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "4",
+   "metadata": {},
+   "source": [
+    "\n",
+    "# Resiliency\n",
+    "\n",
+    "Scenarios can run for a long time, and because of that, things can go wrong. Network issues, rate limits, or other transient failures can interrupt execution. PyRIT provides built-in resiliency features to handle these situations gracefully.\n",
+    "\n",
+    "## Automatic Resume\n",
+    "\n",
+    "If you re-run a `scenario`, it will automatically start where it left off. The framework tracks completed attacks and objectives in memory, so you won't lose progress if something interrupts your scenario execution. This means you can safely stop and restart scenarios without duplicating work.\n",
+    "\n",
+    "## Retry Mechanism\n",
+    "\n",
+    "You can utilize the `max_retries` parameter to handle transient failures. If any unknown exception occurs during execution, PyRIT will automatically retry the failed operation (starting where it left off) up to the specified number of times. This helps ensure your scenario completes successfully even in the face of temporary issues.\n",
+    "\n",
+    "## Dynamic Configuration\n",
+    "\n",
+    "During a long-running scenario, you may want to adjust parameters like `max_concurrency` to manage resource usage, or switch your scorer to use a different target. PyRIT's resiliency features make it safe to stop, reconfigure, and continue scenarios as needed.\n",
+    "\n",
+    "For more information, see [resiliency](../setup/2_resiliency.ipynb)"
+   ]
+  }
+ ],
+ "metadata": {
+  "language_info": {
+   "codemirror_mode": {
+    "name": "ipython",
+    "version": 3
+   },
+   "file_extension": ".py",
+   "mimetype": "text/x-python",
+   "name": "python",
+   "nbconvert_exporter": "python",
+   "pygments_lexer": "ipython3",
+   "version": "3.11.14"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 5
+}