diff --git a/doc/scanner/airt.ipynb b/doc/scanner/airt.ipynb index 7a1e2cf1eb..d088760e7f 100644 --- a/doc/scanner/airt.ipynb +++ b/doc/scanner/airt.ipynb @@ -34,29 +34,9 @@ "text": [ "Found default environment files: ['./.pyrit/.env', './.pyrit/.env.local']\n", "Loaded environment file: ./.pyrit/.env\n", - "Loaded environment file: ./.pyrit/.env.local\n" - ] - }, - { - "name": "stdout", - "output_type": "stream", - "text": [ + "Loaded environment file: ./.pyrit/.env.local\n", "[pyrit:alembic] No new upgrade operations detected.\n" ] - }, - { - "name": "stderr", - "output_type": "stream", - "text": [ - "TextAdaptive: _EXCLUDED_TECHNIQUES entries ['prompt_sending'] are not in the current scenario-techniques catalog ['context_compliance', 'crescendo_history_lecture', 'crescendo_journalist_interview', 'crescendo_movie_director', 'crescendo_simulated', 'flip', 'many_shot', 'pair', 'red_teaming', 'role_play_movie_script', 'role_play_persuasion', 'role_play_persuasion_written', 'role_play_trivia_game', 'role_play_video_game', 'tap', 'violent_durian']; the exclusion is a no-op for those entries. Remove stale entries or update the catalog.\n" - ] - }, - { - "name": "stderr", - "output_type": "stream", - "text": [ - "\n" - ] } ], "source": [ @@ -71,9 +51,23 @@ " TechniqueInitializer,\n", ")\n", "\n", + "dataset_initializer = LoadDefaultDatasets()\n", + "dataset_initializer.set_params_from_args(\n", + " args={\n", + " \"dataset_names\": [\n", + " \"airt_hate\",\n", + " \"airt_imminent_crisis\",\n", + " \"airt_leakage\",\n", + " \"airt_malware\",\n", + " \"airt_scams\",\n", + " \"harmbench\",\n", + " ]\n", + " }\n", + ")\n", + "\n", "await initialize_pyrit_async( # type: ignore\n", " memory_db_type=IN_MEMORY,\n", - " initializers=[TargetInitializer(), ScorerInitializer(), TechniqueInitializer(), LoadDefaultDatasets()],\n", + " initializers=[TargetInitializer(), ScorerInitializer(), TechniqueInitializer(), dataset_initializer],\n", ")\n", "\n", "objective_target = OpenAIChatTarget()" @@ -1267,6 +1261,163 @@ "cell_type": "markdown", "id": "15", "metadata": {}, + "source": [ + "## Multilingual\n", + "\n", + "Tests whether target safeguards remain effective when harmful objectives are presented in other\n", + "languages. A run crosses registered text-compatible attack techniques with datasets and translation\n", + "strategies. By default, `translation` translates each objective into every selected language, and\n", + "`random_translation` translates individual words using the full selected language pool. A baseline\n", + "sends each objective without translation and is included by default.\n", + "\n", + "```bash\n", + "pyrit_scan airt.multilingual \\\n", + " --initializers target load_default_datasets \\\n", + " --target openai_chat \\\n", + " --dataset-names harmbench \\\n", + " --max-dataset-size 1\n", + "```\n", + "\n", + "**Available techniques:** `prompt_sending` is the default. Every registry technique (`role_play_*`,\n", + "`many_shot`, `tap`, …) whose built-in request converter chain ends in text is also available.\n", + "\n", + "**Translation strategies:** `translation` and `random_translation` (both default). A bare run translates\n", + "five objectives into five randomly selected languages, plus a word-level random language translation.\n", + "Pass `num_languages` to change the random sample size or `languages` to provide an explicit list.\n", + "The two language selectors are mutually exclusive." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "16", + "metadata": {}, + "outputs": [ + { + "data": { + "application/vnd.jupyter.widget-view+json": { + "model_id": "f5b9ec4dd8ea441196f7fc1ee4c80305", + "version_major": 2, + "version_minor": 0 + }, + "text/plain": [ + "Executing Multilingual: 0%| | 0/5 [00:00 ComponentIdentifier: + """ + Build the converter identifier with the random translation language pool. + + Returns: + ComponentIdentifier: The converter identifier. + """ + base_identifier = super()._build_identifier() + return self._create_identifier( + params={ + **base_identifier.params, + "languages": sorted(self.languages, key=str.casefold), + }, + converter_target=self._converter_target.get_identifier(), + ) + async def convert_async(self, *, prompt: str, input_type: PromptDataType = "text") -> ConverterResult: """ Convert the given prompt into the target format supported by the converter. diff --git a/pyrit/scenario/core/attack_technique_factory.py b/pyrit/scenario/core/attack_technique_factory.py index 1d168a0545..bf415e4e22 100644 --- a/pyrit/scenario/core/attack_technique_factory.py +++ b/pyrit/scenario/core/attack_technique_factory.py @@ -32,6 +32,7 @@ AttackTechniqueSeedGroup, ComponentIdentifier, Identifiable, + PromptDataType, SeedIdentifier, SeedPrompt, SeedSimulatedConversation, @@ -41,6 +42,7 @@ from pyrit.scenario.core.scenario_target_defaults import get_default_adversarial_target if TYPE_CHECKING: + from pyrit.converter import Converter from pyrit.executor.attack import AttackStrategy from pyrit.prompt_normalizer import ConverterConfiguration from pyrit.prompt_target import PromptTarget @@ -404,6 +406,54 @@ def seed_technique(self) -> AttackTechniqueSeedGroup | None: """The optional technique seed group.""" return self._seed_technique + def can_append_request_converter(self, *, converter_type: type[Converter]) -> bool: + """ + Return whether ``converter_type`` can safely follow the baked request converter chain. + + The factory starts with a text objective and projects the possible output modalities + through each baked request converter. Conditional converter configurations preserve the + unconverted modality as another possible path. The appended converter must accept every + resulting modality, and the attack class must expose ``attack_converter_config`` so the + converter is not silently ignored by ``create()``. + + Args: + converter_type (type[Converter]): The request converter type to append. + + Returns: + bool: ``True`` when the converter can be appended safely. + """ + if "attack_converter_config" not in self._get_accepted_params(): + return False + + output_types: set[PromptDataType] = {"text"} + converter_config = self._attack_kwargs.get("attack_converter_config") + if converter_config is None: + return "text" in converter_type.SUPPORTED_INPUT_TYPES + + for configuration in converter_config.request_converters: + next_output_types: set[PromptDataType] = set() + for output_type in output_types: + applies_to_type = ( + not configuration.prompt_data_types_to_apply + or output_type in configuration.prompt_data_types_to_apply + ) + if not applies_to_type: + next_output_types.add(output_type) + continue + + converted_types: set[PromptDataType] = {output_type} + for built_in_converter in configuration.converters: + if not all(built_in_converter.input_supported(data_type) for data_type in converted_types): + return False + converted_types = set(built_in_converter.supported_output_types) + + next_output_types.update(converted_types) + if configuration.indexes_to_apply: + next_output_types.add(output_type) + output_types = next_output_types + + return bool(output_types) and output_types.issubset(converter_type.SUPPORTED_INPUT_TYPES) + @property def adversarial_chat(self) -> PromptTarget | None: """The adversarial chat target baked into this factory, or None.""" diff --git a/pyrit/scenario/scenarios/airt/__init__.py b/pyrit/scenario/scenarios/airt/__init__.py index 61fdbdfb56..df6770c4d2 100644 --- a/pyrit/scenario/scenarios/airt/__init__.py +++ b/pyrit/scenario/scenarios/airt/__init__.py @@ -8,6 +8,7 @@ from pyrit.scenario.scenarios.airt.cyber import Cyber, _build_cyber_technique from pyrit.scenario.scenarios.airt.jailbreak import Jailbreak, _build_jailbreak_technique from pyrit.scenario.scenarios.airt.leakage import Leakage, _build_leakage_technique +from pyrit.scenario.scenarios.airt.multilingual import Multilingual, _build_multilingual_technique from pyrit.scenario.scenarios.airt.psychosocial import Psychosocial, PsychosocialTechnique from pyrit.scenario.scenarios.airt.rapid_response import RapidResponse, _build_rapid_response_technique from pyrit.scenario.scenarios.airt.scam import Scam, ScamTechnique @@ -31,6 +32,8 @@ def __getattr__(name: str) -> Any: return _build_cyber_technique() if name == "JailbreakTechnique": return _build_jailbreak_technique() + if name == "MultilingualTechnique": + return _build_multilingual_technique() raise AttributeError(f"module {__name__!r} has no attribute {name!r}") @@ -41,6 +44,8 @@ def __getattr__(name: str) -> Any: "JailbreakTechnique", "Leakage", "LeakageTechnique", + "Multilingual", + "MultilingualTechnique", "Psychosocial", "PsychosocialTechnique", "RapidResponse", diff --git a/pyrit/scenario/scenarios/airt/multilingual.py b/pyrit/scenario/scenarios/airt/multilingual.py new file mode 100644 index 0000000000..bc70be31e6 --- /dev/null +++ b/pyrit/scenario/scenarios/airt/multilingual.py @@ -0,0 +1,364 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +from __future__ import annotations + +import logging +import random +from functools import cache +from typing import TYPE_CHECKING, Any, ClassVar, Literal + +from pyrit.common import apply_defaults +from pyrit.common.path import DATASETS_PATH +from pyrit.converter import RandomTranslationConverter, TranslationConverter +from pyrit.executor.attack import PromptSendingAttack +from pyrit.models import Parameter, SeedDataset +from pyrit.registry.components.attack_technique_registry import AttackTechniqueRegistry +from pyrit.scenario.core import ( + AtomicAttack, + AttackTechniqueFactory, + BaselineAttackPolicy, + DatasetAttackConfiguration, + Scenario, + ScenarioTechnique, + get_default_adversarial_target, +) +from pyrit.scenario.core.matrix_atomic_attack_builder import ( + MatrixAtomicAttackBuilder, + build_baseline_atomic_attack, + resolve_technique_factories, +) + +if TYPE_CHECKING: + from pyrit.prompt_target import PromptTarget + from pyrit.scenario.core import ScenarioTechnique + from pyrit.scenario.core.scenario_context import ScenarioContext + from pyrit.score import TrueFalseScorer + +logger = logging.getLogger(__name__) + +# Metadata key under which the resolved languages are persisted, so a resumed run +# replays the exact same set even when a random sample was drawn. +_LANGUAGES_METADATA_KEY = "languages" + +# How many languages a bare run draws at random. Languages multiply against objectives and +# techniques for fixed translation. Override per run with +# ``num_languages`` (random count) or ``languages`` (an explicit set). +_DEFAULT_NUM_LANGUAGES = 5 + +_PROMPT_SENDING = "prompt_sending" +_TRANSLATION = "translation" +_RANDOM_TRANSLATION = "random_translation" +TranslationStrategy = Literal["translation", "random_translation"] + + +def _normalize_languages(languages: list[str]) -> list[str]: + """ + Normalize and deduplicate language names while preserving their first spelling. + + Args: + languages (list[str]): Language names to normalize. + + Returns: + list[str]: Unique normalized language names. + + Raises: + ValueError: If a language name is empty after normalization. + """ + normalized_by_key: dict[str, str] = {} + for language in languages: + normalized = " ".join(language.replace("_", " ").split()) + if not normalized: + raise ValueError("languages must not contain empty language names.") + normalized_by_key.setdefault(normalized.casefold(), normalized) + return list(normalized_by_key.values()) + + +def _language_key(language: str) -> str: + """ + Build the normalized language key used in atomic attack names. + + Args: + language (str): A normalized language name. + + Returns: + str: The lowercase key with spaces replaced by underscores. + """ + return language.casefold().replace(" ", "_") + + +@cache +def _prompt_sending_factory() -> AttackTechniqueFactory: + """ + Build the scenario-local bare prompt-sending technique factory. + + Returns: + AttackTechniqueFactory: The prompt-sending factory. + """ + return AttackTechniqueFactory( + name=_PROMPT_SENDING, + attack_class=PromptSendingAttack, + technique_tags=["single_turn"], + ) + + +def _extra_default_factories() -> dict[str, AttackTechniqueFactory]: + """Return scenario-local technique factories keyed by name.""" + return {_PROMPT_SENDING: _prompt_sending_factory()} + + +@cache +def _build_multilingual_technique() -> type[ScenarioTechnique]: + """ + Build the Multilingual technique class from text-compatible registered factories. + + Returns: + type[ScenarioTechnique]: The dynamically generated technique enum class. + """ + registry = AttackTechniqueRegistry.get_registry_singleton() + factories = [ + factory + for factory in list(registry.get_factories_or_raise().values()) + list(_extra_default_factories().values()) + if factory.can_append_request_converter(converter_type=TranslationConverter) + ] + return AttackTechniqueRegistry.build_technique_class_from_factories( # type: ignore[ty:invalid-return-type] + class_name="MultilingualTechnique", + factories=factories, + default_names={_PROMPT_SENDING}, + ) + + +class Multilingual(Scenario): + """ + Multilingual scenario implementation for PyRIT. + + Tests how vulnerable a model is to non-English language use. + """ + + VERSION: int = 1 + BASELINE_ATTACK_POLICY: ClassVar[BaselineAttackPolicy] = BaselineAttackPolicy.Enabled + + # Default language list + _DEFAULT_LANGUAGES_SEED_PROMPT_PATH = DATASETS_PATH / "lexicons" / "languages_most_spoken.yaml" + + @classmethod + def required_datasets(cls) -> list[str]: + """Return a list of dataset names required by this scenario.""" + return ["harmbench"] + + @classmethod + def additional_parameters(cls) -> list[Parameter]: + """ + Declare the run-configurable parameters this scenario accepts (CLI / config file). + + Returns: + list[Parameter]: The language selectors and translation strategy selector. + """ + return [ + Parameter( + name="num_languages", + description="Draw this many random languages. Mutually exclusive with languages.", + param_type=int, + default=None, + ), + Parameter( + name="languages", + description=( + "Explicit languages to use (e.g. French, German, Spanish). " + "When omitted, a random sample is drawn. Mutually exclusive with num_languages." + ), + param_type=list[str], + default=None, + ), + Parameter( + name="translation_strategies", + description=( + "Translation strategies to run: translation translates the complete objective into each " + "selected language; random_translation translates words using the selected language pool." + ), + param_type=list[TranslationStrategy], + default=[_TRANSLATION, _RANDOM_TRANSLATION], + ), + ] + + @classmethod + def supported_parameters(cls) -> list[Parameter]: + """ + Declare supported inputs, excluding user-supplied technique converters. + + Returns: + list[Parameter]: The supported scenario parameters. + """ + return [parameter for parameter in super().supported_parameters() if parameter.name != "technique_converters"] + + @apply_defaults + def __init__( + self, + *, + adversarial_chat: PromptTarget | None = None, + objective_scorer: TrueFalseScorer | None = None, + scenario_result_id: str | None = None, + ) -> None: + """ + Initialize the multilingual scenario. + + Args: + adversarial_chat (PromptTarget | None): Target used by the translation converters. + objective_scorer (TrueFalseScorer | None): Scorer used to evaluate target responses. + scenario_result_id (str | None): Optional ID of an existing scenario result to resume. + """ + self._adversarial_chat = adversarial_chat + self._objective_scorer: TrueFalseScorer = ( + objective_scorer if objective_scorer else self._get_default_objective_scorer() + ) + self._default_languages = self._get_default_languages() + self._resolved_languages: list[str] = [] + + technique_class = _build_multilingual_technique() + + super().__init__( + version=self.VERSION, + technique_class=technique_class, + default_dataset_config=DatasetAttackConfiguration(dataset_names=["harmbench"], max_dataset_size=5), + objective_scorer=self._objective_scorer, + scenario_result_id=scenario_result_id, + ) + + @classmethod + def _get_default_languages(cls) -> list[str]: + """ + Load the default languages from the public PyRIT lexicon. + + Returns: + list[str]: The list of most-spoken languages. + """ + dataset = SeedDataset.from_yaml_file(cls._DEFAULT_LANGUAGES_SEED_PROMPT_PATH) + return [str(seed.value) for seed in dataset.seeds] + + def _resolve_languages(self) -> list[str]: + """ + Resolve the languages for this run, replaying the persisted set on resume. + + On a fresh run this reads the run parameters: an explicit ``languages`` set or a random + ``num_languages`` sample (defaulting to a small random draw when neither is given). On resume + the originally chosen set is read back from the stored ``ScenarioResult`` metadata so a random + sample isn't redrawn (which would diverge from the persisted attacks). + + Returns: + list[str]: The explicit or randomly sampled languages for this run. + + Raises: + ValueError: If both ``num_languages`` and ``languages`` are provided, + or if ``num_languages`` is out of bounds. + """ + if self._scenario_result_id is not None: + stored = self._memory.get_scenario_results(scenario_result_ids=[self._scenario_result_id]) + if stored: + persisted = (stored[0].metadata or {}).get(_LANGUAGES_METADATA_KEY) + if persisted: + return _normalize_languages(list(persisted)) + + num_languages = self.params.get("num_languages") + languages = self.params.get("languages") + + if num_languages is not None and languages is not None: + raise ValueError( + "Please provide only one of `num_languages` (random selection) or `languages` (specific selection)." + ) + + if languages is not None: + if not languages: + raise ValueError("languages must contain at least one language.") + return _normalize_languages(languages) + + count = int(num_languages) if num_languages is not None else _DEFAULT_NUM_LANGUAGES + if count < 1 or count > len(self._default_languages): + raise ValueError(f"num_languages must be between 1 and {len(self._default_languages)}.") + return _normalize_languages(random.sample(self._default_languages, count)) + + def _build_initial_scenario_metadata(self) -> dict[str, Any]: + """ + Persist the resolved languages alongside the base scenario metadata. + + Returns: + dict[str, Any]: The base metadata plus the resolved language set. + """ + metadata = super()._build_initial_scenario_metadata() + metadata[_LANGUAGES_METADATA_KEY] = list(self._resolved_languages) + return metadata + + async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list[AtomicAttack]: + """ + Build the technique x dataset x translation-strategy/language attack matrix. + + Args: + context (ScenarioContext): The resolved runtime inputs for this run. + + Returns: + list[AtomicAttack]: The atomic attacks to execute. + + Raises: + ValueError: If the scenario is not properly initialized. + """ + if self._objective_target is None: + raise ValueError( + "Scenario not properly initialized. Call await scenario.initialize_async() before running." + ) + + self._resolved_languages = self._resolve_languages() + adversarial_chat = self._adversarial_chat or get_default_adversarial_target() + strategies = set(self.params.get("translation_strategies") or [_TRANSLATION, _RANDOM_TRANSLATION]) + technique_factories = resolve_technique_factories( + context=context, + extra_factories=_extra_default_factories(), + ) + builder = MatrixAtomicAttackBuilder( + objective_target=context.objective_target, + objective_scorer=self._objective_scorer, + memory_labels=context.memory_labels, + ) + + atomic_attacks: list[AtomicAttack] = [] + if context.include_baseline: + atomic_attacks.append( + build_baseline_atomic_attack( + objective_target=context.objective_target, + objective_scorer=self._objective_scorer, + seed_groups=list(context.seed_groups), + memory_labels=context.memory_labels, + ) + ) + + if _TRANSLATION in strategies: + for language in self._resolved_languages: + converter = TranslationConverter(converter_target=adversarial_chat, language=language) + atomic_attacks.extend( + builder.build( + technique_factories=technique_factories, + dataset_groups=context.seed_groups_by_dataset, + technique_converters={name: [converter] for name in technique_factories}, + name_fn=lambda combo, language=language: ( + f"{combo.technique_name}_{_TRANSLATION}_{_language_key(language)}_{combo.dataset_name}" + ), + display_group_fn=lambda combo, language=language: language, + include_baseline=False, + ) + ) + + if _RANDOM_TRANSLATION in strategies: + converter = RandomTranslationConverter( + converter_target=adversarial_chat, + languages=self._resolved_languages, + ) + atomic_attacks.extend( + builder.build( + technique_factories=technique_factories, + dataset_groups=context.seed_groups_by_dataset, + technique_converters={name: [converter] for name in technique_factories}, + name_fn=lambda combo: f"{combo.technique_name}_{_RANDOM_TRANSLATION}_{combo.dataset_name}", + display_group_fn=lambda combo: "Random Translation", + include_baseline=False, + ) + ) + + return atomic_attacks diff --git a/tests/unit/converter/test_random_translation_converter.py b/tests/unit/converter/test_random_translation_converter.py index 652610792c..5a21c34238 100644 --- a/tests/unit/converter/test_random_translation_converter.py +++ b/tests/unit/converter/test_random_translation_converter.py @@ -56,3 +56,17 @@ def test_random_translation_converter_custom_languages() -> None: assert len(converter.languages) == 3 assert "French" in converter.languages assert "Javanese" not in converter.languages + + +def test_random_translation_converter_identifier_canonicalizes_language_order(mock_target) -> None: + first = RandomTranslationConverter(converter_target=mock_target, languages=["French", "Spanish"]) + reordered = RandomTranslationConverter(converter_target=mock_target, languages=["Spanish", "French"]) + + assert first.get_identifier().hash == reordered.get_identifier().hash + + +def test_random_translation_converter_identifier_distinguishes_language_pools(mock_target) -> None: + first = RandomTranslationConverter(converter_target=mock_target, languages=["French", "Spanish"]) + different = RandomTranslationConverter(converter_target=mock_target, languages=["German", "Japanese"]) + + assert first.get_identifier().hash != different.get_identifier().hash diff --git a/tests/unit/scenario/airt/test_multilingual.py b/tests/unit/scenario/airt/test_multilingual.py new file mode 100644 index 0000000000..c5b238d4d2 --- /dev/null +++ b/tests/unit/scenario/airt/test_multilingual.py @@ -0,0 +1,337 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +"""Tests for the Multilingual scenario.""" + +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from pyrit.converter import Base64Converter, QRCodeConverter, RandomTranslationConverter, TranslationConverter +from pyrit.executor.attack import AttackConverterConfig, PromptSendingAttack +from pyrit.models import AttackSeedGroup, ComponentIdentifier, SeedObjective +from pyrit.prompt_normalizer import ConverterConfiguration +from pyrit.prompt_target import PromptTarget +from pyrit.registry.components.attack_technique_registry import AttackTechniqueRegistry +from pyrit.scenario.core.attack_technique_factory import AttackTechniqueFactory +from pyrit.scenario.scenarios.airt.multilingual import ( + _DEFAULT_NUM_LANGUAGES, + _LANGUAGES_METADATA_KEY, + _PROMPT_SENDING, + _RANDOM_TRANSLATION, + _TRANSLATION, + Multilingual, + _build_multilingual_technique, +) +from pyrit.score import TrueFalseScorer + + +def _mock_identifier(name: str) -> ComponentIdentifier: + """Build a component identifier for a mock scenario dependency.""" + return ComponentIdentifier(class_name=name, class_module="test") + + +@pytest.fixture(autouse=True) +def reset_technique_registry(): + """Register one compatible and one incompatible technique for catalog tests.""" + AttackTechniqueRegistry.reset_registry_singleton() + _build_multilingual_technique.cache_clear() + + text_factory = AttackTechniqueFactory( + name="base64", + attack_class=PromptSendingAttack, + technique_tags=["single_turn"], + attack_kwargs={ + "attack_converter_config": AttackConverterConfig( + request_converters=ConverterConfiguration.from_converters(converters=[Base64Converter()]) + ) + }, + ) + image_factory = AttackTechniqueFactory( + name="qr_code", + attack_class=PromptSendingAttack, + technique_tags=["single_turn"], + attack_kwargs={ + "attack_converter_config": AttackConverterConfig( + request_converters=ConverterConfiguration.from_converters(converters=[QRCodeConverter()]) + ) + }, + ) + AttackTechniqueRegistry.get_registry_singleton().register_from_factories([text_factory, image_factory]) + yield + AttackTechniqueRegistry.reset_registry_singleton() + _build_multilingual_technique.cache_clear() + + +@pytest.fixture +def mock_memory_seed_groups() -> list[AttackSeedGroup]: + """Create an inline objective population.""" + return [AttackSeedGroup(seeds=[SeedObjective(value="test objective")])] + + +@pytest.fixture +def mock_objective_target() -> PromptTarget: + """Create the target under test.""" + mock = MagicMock(spec=PromptTarget) + mock.get_identifier.return_value = _mock_identifier("MockObjectiveTarget") + mock.configuration.includes.return_value = True + return mock + + +@pytest.fixture +def mock_adversarial_chat() -> PromptTarget: + """Create the target used by translation converters.""" + mock = MagicMock(spec=PromptTarget) + mock.get_identifier.return_value = _mock_identifier("MockAdversarialChat") + mock.capabilities.includes.return_value = True + return mock + + +@pytest.fixture +def mock_objective_scorer() -> TrueFalseScorer: + """Create the objective scorer.""" + mock = MagicMock(spec=TrueFalseScorer) + mock.get_identifier.return_value = _mock_identifier("MockObjectiveScorer") + return mock + + +def _patch_seed_groups(mock_memory_seed_groups): + return patch.object( + Multilingual, + "_resolve_seed_groups_by_dataset_async", + new_callable=AsyncMock, + return_value={"harmbench": mock_memory_seed_groups}, + ) + + +def _request_converters(atomic_attack): + """Return the flattened request converter chain configured on an atomic attack.""" + configurations = atomic_attack.attack_technique.attack.get_request_converters() + return [converter for configuration in configurations for converter in configuration.converters] + + +@pytest.mark.usefixtures("patch_central_database") +class TestMultilingual: + """Validate multilingual technique selection and converter construction.""" + + def test_technique_catalog_includes_only_translation_compatible_factories(self) -> None: + technique_class = _build_multilingual_technique() + + all_values = {technique.value for technique in technique_class.expand({technique_class.ALL})} + default_values = {technique.value for technique in technique_class.expand({technique_class.default()})} + assert all_values == {_PROMPT_SENDING, "base64"} + assert default_values == {_PROMPT_SENDING} + + def test_declares_run_parameters(self) -> None: + """Language and strategy selectors are declared while user converter stacks are rejected.""" + parameters = {parameter.name: parameter for parameter in Multilingual.additional_parameters()} + supported_names = {parameter.name for parameter in Multilingual.supported_parameters()} + assert set(parameters) == {"num_languages", "languages", "translation_strategies"} + assert set(parameters).issubset(supported_names) + assert "technique_converters" not in supported_names + assert set(parameters["translation_strategies"].choices or []) == {_TRANSLATION, _RANDOM_TRANSLATION} + + async def test_default_draws_five_random_languages( + self, mock_objective_target, mock_adversarial_chat, mock_objective_scorer, mock_memory_seed_groups + ): + selected = ["French", "Spanish"] + + with ( + _patch_seed_groups(mock_memory_seed_groups), + patch("pyrit.scenario.scenarios.airt.multilingual.random.sample", return_value=selected) as sample, + ): + scenario = Multilingual( + adversarial_chat=mock_adversarial_chat, + objective_scorer=mock_objective_scorer, + ) + scenario.set_params_from_args(args={"objective_target": mock_objective_target}) + await scenario.initialize_async() + assert scenario._resolved_languages == selected + assert sample.call_args.args[1] == _DEFAULT_NUM_LANGUAGES + + async def test_num_languages_samples_that_many( + self, mock_objective_target, mock_adversarial_chat, mock_objective_scorer, mock_memory_seed_groups + ): + selected = ["French", "German", "Spanish"] + + with ( + _patch_seed_groups(mock_memory_seed_groups), + patch("pyrit.scenario.scenarios.airt.multilingual.random.sample", return_value=selected) as sample, + ): + scenario = Multilingual( + adversarial_chat=mock_adversarial_chat, + objective_scorer=mock_objective_scorer, + ) + scenario.set_params_from_args(args={"objective_target": mock_objective_target, "num_languages": 3}) + await scenario.initialize_async() + assert scenario._resolved_languages == selected + assert sample.call_args.args[1] == 3 + + async def test_both_translation_strategies_build_distinct_matrix_slices( + self, mock_objective_target, mock_adversarial_chat, mock_objective_scorer, mock_memory_seed_groups + ): + with _patch_seed_groups(mock_memory_seed_groups): + scenario = Multilingual( + adversarial_chat=mock_adversarial_chat, + objective_scorer=mock_objective_scorer, + ) + scenario.set_params_from_args( + args={ + "objective_target": mock_objective_target, + "languages": ["Canadian French", "Spanish"], + "translation_strategies": [_TRANSLATION, _RANDOM_TRANSLATION], + } + ) + await scenario.initialize_async() + + assert [attack.atomic_attack_name for attack in scenario._atomic_attacks] == [ + "baseline", + "prompt_sending_translation_canadian_french_harmbench", + "prompt_sending_translation_spanish_harmbench", + "prompt_sending_random_translation_harmbench", + ] + converters = [_request_converters(attack) for attack in scenario._atomic_attacks[1:4]] + assert all(isinstance(converter_chain[-1], TranslationConverter) for converter_chain in converters[0:2]) + assert isinstance(converters[2][-1], RandomTranslationConverter) + assert converters[2][-1].languages == ["Canadian French", "Spanish"] + + async def test_registered_technique_preserves_built_in_converter_before_translation( + self, mock_objective_target, mock_adversarial_chat, mock_objective_scorer, mock_memory_seed_groups + ): + technique_class = _build_multilingual_technique() + with _patch_seed_groups(mock_memory_seed_groups): + scenario = Multilingual( + adversarial_chat=mock_adversarial_chat, + objective_scorer=mock_objective_scorer, + ) + scenario.set_params_from_args( + args={ + "objective_target": mock_objective_target, + "scenario_techniques": [technique_class.base64], + "languages": ["French"], + "include_baseline": False, + } + ) + await scenario.initialize_async() + + converters = _request_converters(scenario._atomic_attacks[0]) + assert [type(converter) for converter in converters] == [Base64Converter, TranslationConverter] + + async def test_language_normalization_preserves_unique_atomic_attack_names( + self, mock_objective_target, mock_adversarial_chat, mock_objective_scorer, mock_memory_seed_groups + ): + with _patch_seed_groups(mock_memory_seed_groups): + scenario = Multilingual( + adversarial_chat=mock_adversarial_chat, + objective_scorer=mock_objective_scorer, + ) + scenario.set_params_from_args( + args={ + "objective_target": mock_objective_target, + "languages": ["Canadian French", "canadian_french", " SPANISH ", "spanish"], + "translation_strategies": [_TRANSLATION], + "include_baseline": False, + } + ) + await scenario.initialize_async() + + assert scenario._resolved_languages == ["Canadian French", "SPANISH"] + assert [attack.atomic_attack_name for attack in scenario._atomic_attacks] == [ + "prompt_sending_translation_canadian_french_harmbench", + "prompt_sending_translation_spanish_harmbench", + ] + + async def test_mutually_exclusive_selectors_raise( + self, mock_objective_target, mock_adversarial_chat, mock_objective_scorer, mock_memory_seed_groups + ): + with _patch_seed_groups(mock_memory_seed_groups): + scenario = Multilingual( + adversarial_chat=mock_adversarial_chat, + objective_scorer=mock_objective_scorer, + ) + scenario.set_params_from_args( + args={ + "objective_target": mock_objective_target, + "num_languages": 2, + "languages": ["French"], + } + ) + with pytest.raises(ValueError, match="only one of"): + await scenario.initialize_async() + + def test_invalid_translation_strategy_raises(self, mock_adversarial_chat, mock_objective_scorer): + scenario = Multilingual( + adversarial_chat=mock_adversarial_chat, + objective_scorer=mock_objective_scorer, + ) + with pytest.raises(ValueError, match="expected one of"): + scenario.set_params_from_args(args={"translation_strategies": ["unknown"]}) + + async def test_metadata_records_resolved_languages( + self, mock_objective_target, mock_adversarial_chat, mock_objective_scorer, mock_memory_seed_groups + ): + with _patch_seed_groups(mock_memory_seed_groups): + scenario = Multilingual( + adversarial_chat=mock_adversarial_chat, + objective_scorer=mock_objective_scorer, + ) + scenario.set_params_from_args( + args={ + "objective_target": mock_objective_target, + "languages": ["French", "Spanish"], + } + ) + await scenario.initialize_async() + + metadata = scenario._build_initial_scenario_metadata() + assert metadata[_LANGUAGES_METADATA_KEY] == ["French", "Spanish"] + + def test_resolve_languages_replays_persisted_set_on_resume(self, mock_adversarial_chat, mock_objective_scorer): + scenario = Multilingual( + adversarial_chat=mock_adversarial_chat, + objective_scorer=mock_objective_scorer, + scenario_result_id="existing-result", + ) + stored = MagicMock() + stored.metadata = {_LANGUAGES_METADATA_KEY: ["French", "Spanish"]} + + with patch.object(scenario._memory, "get_scenario_results", return_value=[stored]): + assert scenario._resolve_languages() == ["French", "Spanish"] + + async def test_baseline_is_prepended_by_default_with_same_seed_population( + self, mock_objective_target, mock_adversarial_chat, mock_objective_scorer, mock_memory_seed_groups + ): + with _patch_seed_groups(mock_memory_seed_groups): + scenario = Multilingual( + adversarial_chat=mock_adversarial_chat, + objective_scorer=mock_objective_scorer, + ) + scenario.set_params_from_args(args={"objective_target": mock_objective_target, "languages": ["French"]}) + await scenario.initialize_async() + + assert scenario._atomic_attacks[0].atomic_attack_name == "baseline" + assert scenario._atomic_attacks[0].seed_groups == scenario._atomic_attacks[1].seed_groups + assert scenario._atomic_attacks[0].seed_groups[0] is scenario._atomic_attacks[1].seed_groups[0] + + async def test_random_translation_pool_changes_technique_evaluation_hash( + self, mock_objective_target, mock_adversarial_chat, mock_objective_scorer, mock_memory_seed_groups + ): + hashes = [] + for languages in (["French", "Spanish"], ["German", "Japanese"]): + with _patch_seed_groups(mock_memory_seed_groups): + scenario = Multilingual( + adversarial_chat=mock_adversarial_chat, + objective_scorer=mock_objective_scorer, + ) + scenario.set_params_from_args( + args={ + "objective_target": mock_objective_target, + "languages": languages, + "translation_strategies": [_RANDOM_TRANSLATION], + "include_baseline": False, + } + ) + await scenario.initialize_async() + hashes.append(scenario._atomic_attacks[0].technique_eval_hash) + + assert hashes[0] != hashes[1] diff --git a/tests/unit/scenario/core/test_attack_technique_factory.py b/tests/unit/scenario/core/test_attack_technique_factory.py index 39e3046685..1124cd4df2 100644 --- a/tests/unit/scenario/core/test_attack_technique_factory.py +++ b/tests/unit/scenario/core/test_attack_technique_factory.py @@ -7,7 +7,7 @@ import pytest -from pyrit.converter import Base64Converter, ROT13Converter +from pyrit.converter import Base64Converter, QRCodeConverter, ROT13Converter, TranslationConverter from pyrit.executor.attack.core.attack_config import AttackConverterConfig, AttackScoringConfig from pyrit.executor.attack.single_turn.prompt_sending import PromptSendingAttack from pyrit.models import AttackTechniqueSeedGroup, ComponentIdentifier, Identifiable, SeedPrompt @@ -178,6 +178,34 @@ def test_validate_kwargs_rejects_invalid_param_on_real_attack_class(self): attack_kwargs={"nonexistent_param": 42}, ) + @pytest.mark.parametrize("baked_converter", [None, Base64Converter()]) + def test_can_append_request_converter_to_text_chain(self, baked_converter): + attack_kwargs = {} + if baked_converter: + attack_kwargs["attack_converter_config"] = AttackConverterConfig( + request_converters=ConverterConfiguration.from_converters(converters=[baked_converter]) + ) + factory = AttackTechniqueFactory( + name="test", + attack_class=PromptSendingAttack, + attack_kwargs=attack_kwargs, + ) + + assert factory.can_append_request_converter(converter_type=TranslationConverter) + + def test_cannot_append_text_converter_to_image_chain(self): + factory = AttackTechniqueFactory( + name="test", + attack_class=PromptSendingAttack, + attack_kwargs={ + "attack_converter_config": AttackConverterConfig( + request_converters=ConverterConfiguration.from_converters(converters=[QRCodeConverter()]) + ) + }, + ) + + assert not factory.can_append_request_converter(converter_type=TranslationConverter) + class TestFactoryCreate: """Tests for AttackTechniqueFactory.create()."""