diff --git a/doc/scanner/garak.ipynb b/doc/scanner/garak.ipynb index 3e7fcd807e..5ec637fbf1 100644 --- a/doc/scanner/garak.ipynb +++ b/doc/scanner/garak.ipynb @@ -91,6 +91,11 @@ "QuotedPrintable, UUencode, ROT13, Braille, Atbash, MorseCode, NATO, Ecoji, Zalgo, LeetSpeak,\n", "AsciiSmuggler\n", "\n", + "**Aggregate techniques:** `ALL` (every encoding, exhaustive) and `DEFAULT` (a broad curated subset\n", + "spanning every encoding family — base-N, byte-encodings, substitution ciphers, and symbolic\n", + "alphabets — for a meaningful default scan; the niche/lossy schemes are ALL-only). `DEFAULT` is used\n", + "when no techniques are specified.\n", + "\n", "> **Note:** Technique composition is NOT supported for Encoding — each encoding is tested\n", "> independently." ] @@ -100,30 +105,7 @@ "execution_count": null, "id": "3", "metadata": {}, - "outputs": [ - { - "name": "stdout", - "output_type": "stream", - "text": [ - "Scenario: Encoding\n", - "Atomic attacks: 21\n" - ] - }, - { - "data": { - "application/vnd.jupyter.widget-view+json": { - "model_id": "e21db36b5f40443593057efcb1ff2588", - "version_major": 2, - "version_minor": 0 - }, - "text/plain": [ - "Executing Encoding: 0%| | 0/21 [00:00 **Note:** Technique composition is NOT supported for Encoding — each encoding is tested # > independently. diff --git a/pyrit/scenario/scenarios/garak/encoding.py b/pyrit/scenario/scenarios/garak/encoding.py index 34dcc8471b..e4b23e13ac 100644 --- a/pyrit/scenario/scenarios/garak/encoding.py +++ b/pyrit/scenario/scenarios/garak/encoding.py @@ -77,33 +77,51 @@ class EncodingTechnique(ScenarioTechnique): Techniques for encoding attacks. Each enum member represents an encoding scheme that will be tested against the target model. - The ALL aggregate expands to include all encoding techniques. + The ``ALL`` aggregate expands to every encoding scheme (exhaustive run). The ``DEFAULT`` + aggregate expands to a curated majority of the schemes (spanning every encoding family), giving + a meaningful default scan; ``ALL`` adds the remaining niche/lossy schemes. Note: EncodingTechnique does not support composition. Each encoding must be applied individually. """ - # Aggregate member + # Aggregate members ALL = ("all", {"all"}) - - # Individual encoding techniques (matching the atomic attack names) - Base64 = ("base64", set[str]()) - Base2048 = ("base2048", set[str]()) - Base16 = ("base16", set[str]()) - Base32 = ("base32", set[str]()) - ASCII85 = ("ascii85", set[str]()) - Hex = ("hex", set[str]()) - QuotedPrintable = ("quoted_printable", set[str]()) - UUencode = ("uuencode", set[str]()) - ROT13 = ("rot13", set[str]()) + DEFAULT = ("default", {"default"}) + + # Individual encoding techniques (the enum value is the encoding scheme name). Members tagged ``default`` + # form the curated DEFAULT aggregate: a broad spread across encoding families so the default scan is a + # meaningful scanner run — base-N (Base64, Base2048, Base16, Base32, ASCII85, Hex), byte-encodings + # (QuotedPrintable, UUencode), substitution ciphers (ROT13, Atbash, LeetSpeak), and symbolic alphabets + # (MorseCode, NATO). The remaining niche/lossy schemes (Braille, Ecoji, Zalgo, AsciiSmuggler) are + # ALL-only. + Base64 = ("base64", {"default"}) + Base2048 = ("base2048", {"default"}) + Base16 = ("base16", {"default"}) + Base32 = ("base32", {"default"}) + ASCII85 = ("ascii85", {"default"}) + Hex = ("hex", {"default"}) + QuotedPrintable = ("quoted_printable", {"default"}) + UUencode = ("uuencode", {"default"}) + ROT13 = ("rot13", {"default"}) Braille = ("braille", set[str]()) - Atbash = ("atbash", set[str]()) - MorseCode = ("morse_code", set[str]()) - NATO = ("nato", set[str]()) + Atbash = ("atbash", {"default"}) + MorseCode = ("morse_code", {"default"}) + NATO = ("nato", {"default"}) Ecoji = ("ecoji", set[str]()) Zalgo = ("zalgo", set[str]()) - LeetSpeak = ("leet_speak", set[str]()) + LeetSpeak = ("leet_speak", {"default"}) AsciiSmuggler = ("ascii_smuggler", set[str]()) + @classmethod + def get_aggregate_tags(cls) -> set[str]: + """ + Get the set of tags that represent aggregate categories. + + Returns: + set[str]: The base ``"all"`` aggregate plus the scenario-specific ``"default"`` aggregate. + """ + return super().get_aggregate_tags() | {"default"} + logger = logging.getLogger(__name__) @@ -125,7 +143,7 @@ class Encoding(Scenario): By default, this uses the same dataset as Garak: slur terms and web XSS payloads. """ - VERSION: int = 1 + VERSION: int = 2 @apply_defaults def __init__( @@ -154,11 +172,11 @@ def __init__( super().__init__( version=self.VERSION, technique_class=EncodingTechnique, - default_technique=EncodingTechnique.ALL, + default_technique=EncodingTechnique.DEFAULT, default_dataset_config=CompoundDatasetAttackConfiguration( configurations=[ - EncodingDatasetConfiguration(dataset_names=["garak_slur_terms_en"], max_dataset_size=3), - EncodingDatasetConfiguration(dataset_names=["garak_web_html_js"], max_dataset_size=3), + EncodingDatasetConfiguration(dataset_names=["garak_slur_terms_en"], max_dataset_size=10), + EncodingDatasetConfiguration(dataset_names=["garak_web_html_js"], max_dataset_size=10), ] ), objective_scorer=objective_scorer, @@ -179,10 +197,10 @@ async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list Returns: list[AtomicAttack]: The list of AtomicAttack instances in this scenario. """ - return self._get_converter_attacks(seed_groups=list(context.seed_groups)) + return self._get_converter_attacks(context=context) # These are the same as Garak encoding attacks - def _get_converter_attacks(self, *, seed_groups: list[AttackSeedGroup]) -> list[AtomicAttack]: + def _get_converter_attacks(self, *, context: ScenarioContext) -> list[AtomicAttack]: """ Get all converter-based atomic attacks. @@ -190,54 +208,68 @@ def _get_converter_attacks(self, *, seed_groups: list[AttackSeedGroup]) -> list[ Each encoding scheme is tested both with and without explicit decoding instructions. Args: - seed_groups (list[AttackSeedGroup]): Seed groups the attacks draw from. + context (ScenarioContext): The resolved runtime inputs for this run. Returns: list[AtomicAttack]: List of all atomic attacks to execute. """ - # Map of all available converters with their encoding names - all_converters_with_encodings: list[tuple[list[Converter], str]] = [ - ([Base64Converter()], "base64"), - ([Base64Converter(encoding_func="urlsafe_b64encode")], "base64"), - ([Base64Converter(encoding_func="standard_b64encode")], "base64"), - ([Base64Converter(encoding_func="b2a_base64")], "base64"), - ([Base2048Converter()], "base2048"), - ([Base64Converter(encoding_func="b16encode")], "base16"), - ([Base64Converter(encoding_func="b32encode")], "base32"), - ([Base64Converter(encoding_func="a85encode")], "ascii85"), - ([Base64Converter(encoding_func="b85encode")], "ascii85"), - ([BinAsciiConverter(encoding_func="hex")], "hex"), - ([BinAsciiConverter(encoding_func="quoted-printable")], "quoted_printable"), - ([BinAsciiConverter(encoding_func="UUencode")], "uuencode"), - ([ROT13Converter()], "rot13"), - ([BrailleConverter()], "braille"), - ([AtbashConverter()], "atbash"), - ([MorseConverter()], "morse_code"), - ([NatoConverter()], "nato"), - ([EcojiConverter()], "ecoji"), - ([ZalgoConverter()], "zalgo"), - ([LeetspeakConverter()], "leet_speak"), - ([AsciiSmugglerConverter()], "ascii_smuggler"), + # Map of all available converters with their encoding name and a unique variant slug. + # ``encoding_name`` drives technique selection and user-facing grouping (display_group); + # ``variant_slug`` is unique per row so atomic-attack names stay unique even when one + # encoding name maps to multiple converter variants (e.g. base64, ascii85). + # NOTE: near-duplicate base64 variants were trimmed alongside the VERSION bump + # (``standard_b64encode`` is byte-identical to the default ``b64encode``; ``b2a_base64`` + # only appends a trailing newline). We keep the default encoding plus the url-safe alphabet, + # which is a genuinely distinct representation. + all_converters_with_encodings: list[tuple[list[Converter], str, str]] = [ + ([Base64Converter()], "base64", "base64"), + ([Base64Converter(encoding_func="urlsafe_b64encode")], "base64", "base64_urlsafe"), + ([Base2048Converter()], "base2048", "base2048"), + ([Base64Converter(encoding_func="b16encode")], "base16", "base16"), + ([Base64Converter(encoding_func="b32encode")], "base32", "base32"), + ([Base64Converter(encoding_func="a85encode")], "ascii85", "ascii85_a85"), + ([Base64Converter(encoding_func="b85encode")], "ascii85", "ascii85_b85"), + ([BinAsciiConverter(encoding_func="hex")], "hex", "hex"), + ([BinAsciiConverter(encoding_func="quoted-printable")], "quoted_printable", "quoted_printable"), + ([BinAsciiConverter(encoding_func="UUencode")], "uuencode", "uuencode"), + ([ROT13Converter()], "rot13", "rot13"), + ([BrailleConverter()], "braille", "braille"), + ([AtbashConverter()], "atbash", "atbash"), + ([MorseConverter()], "morse_code", "morse_code"), + ([NatoConverter()], "nato", "nato"), + ([EcojiConverter()], "ecoji", "ecoji"), + ([ZalgoConverter()], "zalgo", "zalgo"), + ([LeetspeakConverter()], "leet_speak", "leet_speak"), + ([AsciiSmugglerConverter()], "ascii_smuggler", "ascii_smuggler"), ] # Filter to only include selected techniques - selected_encoding_names = {s.value for s in self._scenario_techniques} + selected_encoding_names = {s.value for s in context.scenario_techniques} converters_with_encodings = [ - (conv, name) for conv, name in all_converters_with_encodings if name in selected_encoding_names + (conv, name, variant_slug) + for conv, name, variant_slug in all_converters_with_encodings + if name in selected_encoding_names ] atomic_attacks = [] - for conv, name in converters_with_encodings: + for conv, name, variant_slug in converters_with_encodings: atomic_attacks.extend( - self._get_prompt_attacks(converters=conv, encoding_name=name, seed_groups=seed_groups) + self._get_prompt_attacks( + converters=conv, encoding_name=name, variant_slug=variant_slug, context=context + ) ) return atomic_attacks def _get_prompt_attacks( - self, *, converters: list[Converter], encoding_name: str, seed_groups: list[AttackSeedGroup] + self, + *, + converters: list[Converter], + encoding_name: str, + variant_slug: str, + context: ScenarioContext, ) -> list[AtomicAttack]: """ - Create atomic attacks for a specific encoding scheme. + Create atomic attacks for a specific encoding converter variant. For each seed prompt (the text to be decoded), creates atomic attacks that: 1. Encode the seed prompt using the specified converter(s) @@ -247,43 +279,50 @@ def _get_prompt_attacks( Args: converters (list[Converter]): The list of converters to apply to the seed prompts. - encoding_name (str): Human-readable name of the encoding scheme (e.g., "Base64", "ROT13"). - seed_groups (list[AttackSeedGroup]): Seed groups the attacks draw from. + encoding_name (str): Human-readable name of the encoding scheme (e.g., "base64", "rot13"). + Used as the ``display_group`` so all variants of an encoding aggregate together in output. + variant_slug (str): Unique slug for this converter variant, used to build a unique + ``atomic_attack_name`` per converter variant and prompt config. + context (ScenarioContext): The resolved runtime inputs for this run. Returns: - list[AtomicAttack]: List of atomic attacks for this encoding scheme. - - Raises: - ValueError: If scenario is not properly initialized. + list[AtomicAttack]: List of atomic attacks for this encoding converter variant. """ - converter_configs = [ - AttackConverterConfig(request_converters=ConverterConfiguration.from_converters(converters=converters)) + # (config_name_suffix, converter_config). The bare "raw" config encodes only; each + # decode-template config additionally asks the model to decode. + converter_configs: list[tuple[str, AttackConverterConfig]] = [ + ( + "raw", + AttackConverterConfig(request_converters=ConverterConfiguration.from_converters(converters=converters)), + ) ] - for decode_type in self._encoding_templates: + for decode_index, decode_type in enumerate(self._encoding_templates): converters_ = converters[:] + [AskToDecodeConverter(template=decode_type, encoding_name=encoding_name)] converter_configs.append( - AttackConverterConfig(request_converters=ConverterConfiguration.from_converters(converters=converters_)) + ( + f"decode{decode_index}", + AttackConverterConfig( + request_converters=ConverterConfiguration.from_converters(converters=converters_) + ), + ) ) atomic_attacks = [] - for attack_converter_config in converter_configs: - # objective_target is guaranteed to be non-None by parent class validation - if self._objective_target is None: - raise ValueError( - "Scenario not properly initialized. Call await scenario.initialize_async() before running." - ) + for config_suffix, attack_converter_config in converter_configs: attack = PromptSendingAttack( - objective_target=self._objective_target, + objective_target=context.objective_target, attack_converter_config=attack_converter_config, attack_scoring_config=self._scorer_config, ) atomic_attacks.append( AtomicAttack( - atomic_attack_name=encoding_name, + atomic_attack_name=f"{variant_slug}_{config_suffix}", + display_group=encoding_name, attack_technique=AttackTechnique(attack=attack), - seed_groups=seed_groups, + seed_groups=list(context.seed_groups), + memory_labels=context.memory_labels, ) ) diff --git a/tests/unit/scenario/garak/test_encoding.py b/tests/unit/scenario/garak/test_encoding.py index 50530adef3..eb176f04c1 100644 --- a/tests/unit/scenario/garak/test_encoding.py +++ b/tests/unit/scenario/garak/test_encoding.py @@ -103,7 +103,7 @@ def test_init_with_default_seed_prompts(self, mock_objective_target, mock_object ) assert scenario.name == "Encoding" - assert scenario.VERSION == 1 + assert scenario.VERSION == 2 def test_init_with_custom_scorer(self, mock_objective_target, mock_objective_scorer, mock_memory_seeds): """Test initialization with custom objective scorer.""" @@ -205,12 +205,152 @@ async def test_init_attack_techniques( ) await scenario.initialize_async() - # By default, EncodingTechnique.ALL is used, which expands to all encoding techniques + # By default, EncodingTechnique.DEFAULT is used, which expands to the curated subset assert len(scenario._scenario_techniques) > 0 # Verify all techniques contain EncodingTechnique instances assert all(isinstance(s, EncodingTechnique) for s in scenario._scenario_techniques) - # Verify none of the techniques are the aggregate "ALL" + # Verify none of the techniques are the aggregate members (ALL/DEFAULT) assert all(s != EncodingTechnique.ALL for s in scenario._scenario_techniques) + assert all(s != EncodingTechnique.DEFAULT for s in scenario._scenario_techniques) + # The default run is the curated DEFAULT set, not the exhaustive ALL set + assert {s.value for s in scenario._scenario_techniques} == { + t.value for t in EncodingTechnique.get_techniques_by_tag("default") + } + + +@pytest.mark.usefixtures("patch_central_database") +class TestEncodingTechniqueDefault: + """Tests for the curated DEFAULT aggregate and technique tagging.""" + + def test_default_aggregate_curates_representative_subset(self): + """DEFAULT expands to a broad curated subset spanning encoding families, smaller than ALL.""" + default_names = {t.value for t in EncodingTechnique.get_techniques_by_tag("default")} + assert default_names == { + "base64", + "base2048", + "base16", + "base32", + "ascii85", + "hex", + "quoted_printable", + "uuencode", + "rot13", + "atbash", + "morse_code", + "nato", + "leet_speak", + } + all_names = {t.value for t in EncodingTechnique.get_all_techniques()} + assert default_names < all_names + assert len(all_names) == 17 + # The niche/lossy schemes stay ALL-only. + assert all_names - default_names == {"braille", "ecoji", "zalgo", "ascii_smuggler"} + + def test_get_aggregate_tags_includes_default(self): + """The scenario exposes both ``all`` and ``default`` as aggregate tags.""" + assert EncodingTechnique.get_aggregate_tags() == {"all", "default"} + + def test_default_technique_is_curated_default(self, mock_objective_scorer): + """A bare scenario defaults to the curated DEFAULT aggregate, not ALL.""" + scenario = Encoding(objective_scorer=mock_objective_scorer) + assert scenario._default_technique == EncodingTechnique.DEFAULT + + +@pytest.mark.usefixtures("patch_central_database") +class TestEncodingAtomicNameUniqueness: + """Tests that atomic-attack names are unique per converter variant (collision fix).""" + + async def test_all_atomic_attack_names_are_unique( + self, mock_objective_target, mock_objective_scorer, mock_attack_seed_groups + ): + """Every atomic attack across the exhaustive ALL run has a unique name.""" + from unittest.mock import patch + + with patch.object( + Encoding, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_attack_seed_groups}, + ): + scenario = Encoding(objective_scorer=mock_objective_scorer) + scenario.set_params_from_args( + args={ + "objective_target": mock_objective_target, + "scenario_techniques": [EncodingTechnique.ALL], + } + ) + await scenario.initialize_async() + + names = [ + aa.atomic_attack_name + for aa in scenario._get_converter_attacks( + context=scenario._build_scenario_context(seed_groups_by_dataset={"memory": mock_attack_seed_groups}) + ) + ] + assert len(names) == len(set(names)), "atomic_attack_name collisions detected" + + async def test_base64_trimmed_to_two_variants( + self, mock_objective_target, mock_objective_scorer, mock_attack_seed_groups + ): + """base64 keeps only the default and url-safe variants (near-duplicates removed).""" + from unittest.mock import patch + + with patch.object( + Encoding, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_attack_seed_groups}, + ): + scenario = Encoding(objective_scorer=mock_objective_scorer) + scenario.set_params_from_args( + args={ + "objective_target": mock_objective_target, + "scenario_techniques": [EncodingTechnique.Base64], + } + ) + await scenario.initialize_async() + + attacks = scenario._get_converter_attacks( + context=scenario._build_scenario_context(seed_groups_by_dataset={"memory": mock_attack_seed_groups}) + ) + # Every base64 atomic attack groups under the "base64" display group. + assert all(aa.display_group == "base64" for aa in attacks) + # Two distinct converter variants (default + urlsafe), each fanned over the raw config + # plus one config per decode template — names are unique and prefixed by the variant slug. + names = {aa.atomic_attack_name for aa in attacks} + assert {n for n in names if n.startswith("base64_urlsafe")}, "missing urlsafe variant" + assert {n for n in names if n.startswith("base64_") and not n.startswith("base64_urlsafe")}, ( + "missing default base64 variant" + ) + # The trimmed variants must not appear. + assert not any("standard" in n or "b2a" in n for n in names) + + async def test_memory_labels_propagate_to_atomic_attacks( + self, mock_objective_target, mock_objective_scorer, mock_attack_seed_groups + ): + """Run-level memory_labels reach every built atomic attack (matches the sibling convention).""" + from unittest.mock import patch + + labels = {"experiment": "enc-run", "operator": "airt"} + with patch.object( + Encoding, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"memory": mock_attack_seed_groups}, + ): + scenario = Encoding(objective_scorer=mock_objective_scorer) + scenario.set_params_from_args( + args={ + "objective_target": mock_objective_target, + "scenario_techniques": [EncodingTechnique.ALL], + "memory_labels": labels, + } + ) + await scenario.initialize_async() + + technique_attacks = [aa for aa in scenario._atomic_attacks if aa.atomic_attack_name != "baseline"] + assert technique_attacks + assert all(aa._memory_labels == labels for aa in technique_attacks) @pytest.mark.usefixtures("patch_central_database") @@ -269,10 +409,12 @@ async def test_get_converter_attacks_returns_multiple_encodings( } ) await scenario.initialize_async() - attack_runs = scenario._get_converter_attacks(seed_groups=mock_attack_seed_groups) + attack_runs = scenario._get_converter_attacks( + context=scenario._build_scenario_context(seed_groups_by_dataset={"memory": mock_attack_seed_groups}) + ) # Should have multiple attack runs for different encodings - # The list includes: Base64 (4 variants), Base2048, Base16, Base32, ASCII85 (2), hex, + # The list includes: base64 (2 variants: default + urlsafe), base2048, base16, base32, ascii85 (2), # quoted-printable, UUencode, ROT13, Braille, Atbash, Morse, NATO, Ecoji, Zalgo, Leet, AsciiSmuggler assert len(attack_runs) > 0 @@ -300,7 +442,10 @@ async def test_get_prompt_attacks_creates_attack_runs( ) await scenario.initialize_async() attack_runs = scenario._get_prompt_attacks( - converters=[Base64Converter()], encoding_name="Base64", seed_groups=mock_attack_seed_groups + converters=[Base64Converter()], + encoding_name="base64", + variant_slug="base64", + context=scenario._build_scenario_context(seed_groups_by_dataset={"memory": mock_attack_seed_groups}), ) # Should create attack runs @@ -340,7 +485,10 @@ async def test_attack_runs_include_objectives( ) await scenario.initialize_async() attack_runs = scenario._get_prompt_attacks( - converters=[Base64Converter()], encoding_name="Base64", seed_groups=mock_attack_seed_groups + converters=[Base64Converter()], + encoding_name="base64", + variant_slug="base64", + context=scenario._build_scenario_context(seed_groups_by_dataset={"memory": mock_attack_seed_groups}), ) # Check that seed groups contain objectives with the expected format @@ -428,9 +576,9 @@ def test_default_dataset_config_uses_garak_datasets(self, mock_objective_scorer) assert "garak_web_html_js" in dataset_names def test_default_dataset_config_has_max_size(self, mock_objective_scorer): - """Test that each child of the default config caps samples at 3 per dataset.""" + """Test that each child of the default config caps samples at 10 per dataset.""" config = Encoding(objective_scorer=mock_objective_scorer)._default_dataset_config - assert [child.max_dataset_size for child in config._configurations] == [3, 3] + assert [child.max_dataset_size for child in config._configurations] == [10, 10] @pytest.mark.usefixtures("patch_central_database")