diff --git a/doc/code/scenarios/0_attack_techniques.ipynb b/doc/code/scenarios/0_attack_techniques.ipynb index 556a1fca40..4be7a897f5 100644 --- a/doc/code/scenarios/0_attack_techniques.ipynb +++ b/doc/code/scenarios/0_attack_techniques.ipynb @@ -38,7 +38,13 @@ "\n", "The objective is *not* part of the technique — it stays separate and is supplied by the dataset at\n", "run time. You rarely build a technique by hand; instead you register a **factory** and let scenarios\n", - "construct techniques on demand with the scenario's own objective target and scorer." + "construct techniques on demand with the scenario's own objective target and scorer.\n", + "\n", + "`adversarial_chat_system_prompt` accepts either an inline `SeedPrompt` or a YAML `Path`. The factory\n", + "resolves either representation to a `SeedPrompt` immediately, so created techniques carry portable\n", + "prompt content rather than a runtime file dependency. The legacy\n", + "`adversarial_chat_system_prompt_path` name remains a compatibility alias, and if neither parameter\n", + "is supplied, the existing `red_teaming/{technique_name}.yaml` convention remains the default." ] }, { diff --git a/doc/code/scenarios/0_attack_techniques.py b/doc/code/scenarios/0_attack_techniques.py index 41a149fc7f..711d52d5e9 100644 --- a/doc/code/scenarios/0_attack_techniques.py +++ b/doc/code/scenarios/0_attack_techniques.py @@ -43,6 +43,12 @@ # The objective is *not* part of the technique — it stays separate and is supplied by the dataset at # run time. You rarely build a technique by hand; instead you register a **factory** and let scenarios # construct techniques on demand with the scenario's own objective target and scorer. +# +# `adversarial_chat_system_prompt` accepts either an inline `SeedPrompt` or a YAML `Path`. The factory +# resolves either representation to a `SeedPrompt` immediately, so created techniques carry portable +# prompt content rather than a runtime file dependency. The legacy +# `adversarial_chat_system_prompt_path` name remains a compatibility alias, and if neither parameter +# is supplied, the existing `red_teaming/{technique_name}.yaml` convention remains the default. # %% [markdown] # ## Where techniques come from: initializers diff --git a/pyrit/executor/attack/core/attack_parameters.py b/pyrit/executor/attack/core/attack_parameters.py index e03108448a..e9cb5f886f 100644 --- a/pyrit/executor/attack/core/attack_parameters.py +++ b/pyrit/executor/attack/core/attack_parameters.py @@ -167,6 +167,7 @@ async def from_seed_group_async( num_turns=simulated_conversation_config.num_turns, starting_sequence=simulated_conversation_config.sequence, adversarial_chat_system_prompt_path=simulated_conversation_config.adversarial_chat_system_prompt_path, + adversarial_chat_system_prompt=simulated_conversation_config.adversarial_chat_system_prompt, simulated_target_system_prompt_path=simulated_conversation_config.simulated_target_system_prompt_path, next_message_system_prompt_path=simulated_conversation_config.next_message_system_prompt_path, ) diff --git a/pyrit/executor/attack/multi_turn/simulated_conversation.py b/pyrit/executor/attack/multi_turn/simulated_conversation.py index 872f9571fc..a02926a5dc 100644 --- a/pyrit/executor/attack/multi_turn/simulated_conversation.py +++ b/pyrit/executor/attack/multi_turn/simulated_conversation.py @@ -10,6 +10,7 @@ from __future__ import annotations +import asyncio import logging from typing import TYPE_CHECKING @@ -43,7 +44,8 @@ async def generate_simulated_conversation_async( objective_scorer: TrueFalseScorer, num_turns: int = 3, starting_sequence: int = 0, - adversarial_chat_system_prompt_path: str | Path, + adversarial_chat_system_prompt_path: str | Path | None = None, + adversarial_chat_system_prompt: SeedPrompt | None = None, simulated_target_system_prompt_path: str | Path | None = None, next_message_system_prompt_path: str | Path | None = None, attack_converter_config: AttackConverterConfig | None = None, @@ -70,7 +72,8 @@ async def generate_simulated_conversation_async( num_turns: Number of conversation turns to generate. Defaults to 3. starting_sequence: The starting sequence number for the generated SeedPrompts. Each message gets an incrementing sequence number. Defaults to 0. - adversarial_chat_system_prompt_path: Path to the system prompt for the adversarial chat. + adversarial_chat_system_prompt_path: Legacy path to the system prompt for the adversarial chat. + adversarial_chat_system_prompt: Canonical inline system prompt for the adversarial chat. simulated_target_system_prompt_path: Path to the system prompt for the simulated target. If None, no system prompt is used for the simulated target. next_message_system_prompt_path: Optional path to a system prompt for generating @@ -89,7 +92,7 @@ async def generate_simulated_conversation_async( generated to elicit the objective fulfillment. Raises: - ValueError: If num_turns is not a positive integer. + ValueError: If num_turns is not positive or the adversarial prompt source is ambiguous or missing. """ # Use the same LLM for both adversarial chat and simulated target # They get different system prompts to play different roles @@ -105,10 +108,9 @@ async def generate_simulated_conversation_async( simulated_target_system_prompt_path=simulated_target_system_prompt_path, ) - # Create adversarial config for the simulation. Load the optional path into a SeedPrompt so the - # resolved prompt is stored directly on the configuration. - adversarial_system_prompt = ( - SeedPrompt.from_yaml_file(adversarial_chat_system_prompt_path) if adversarial_chat_system_prompt_path else None + adversarial_system_prompt = await _resolve_adversarial_chat_system_prompt_async( + adversarial_chat_system_prompt_path=adversarial_chat_system_prompt_path, + adversarial_chat_system_prompt=adversarial_chat_system_prompt, ) adversarial_config = AttackAdversarialConfig( target=adversarial_chat, @@ -176,6 +178,35 @@ async def generate_simulated_conversation_async( return seed_prompts +async def _resolve_adversarial_chat_system_prompt_async( + *, + adversarial_chat_system_prompt_path: str | Path | None, + adversarial_chat_system_prompt: SeedPrompt | None, +) -> SeedPrompt: + """ + Adapt a legacy path-backed prompt or canonical inline prompt for execution. + + Args: + adversarial_chat_system_prompt_path: Legacy YAML prompt path. + adversarial_chat_system_prompt: Canonical inline prompt. + + Returns: + The resolved adversarial chat system prompt. + + Raises: + ValueError: If both or neither prompt sources are provided. + """ + has_prompt_path = adversarial_chat_system_prompt_path is not None + has_inline_prompt = adversarial_chat_system_prompt is not None + if has_prompt_path == has_inline_prompt: + raise ValueError("Set exactly one of adversarial_chat_system_prompt_path or adversarial_chat_system_prompt.") + if adversarial_chat_system_prompt is not None: + return adversarial_chat_system_prompt + + assert adversarial_chat_system_prompt_path is not None + return await asyncio.to_thread(SeedPrompt.from_yaml_file, adversarial_chat_system_prompt_path) + + async def _generate_next_message_async( *, objective: str, diff --git a/pyrit/memory/memory_models.py b/pyrit/memory/memory_models.py index 8fc1247bfd..b0797cd337 100644 --- a/pyrit/memory/memory_models.py +++ b/pyrit/memory/memory_models.py @@ -1491,8 +1491,10 @@ def get_seed(self) -> Seed: num_turns=config.get("num_turns", 3), sequence=config.get("sequence", 0), adversarial_chat_system_prompt_path=config.get("adversarial_chat_system_prompt_path"), + adversarial_chat_system_prompt=config.get("adversarial_chat_system_prompt"), simulated_target_system_prompt_path=config.get("simulated_target_system_prompt_path"), next_message_system_prompt_path=config.get("next_message_system_prompt_path"), + pyrit_version=config.get("pyrit_version"), ) return SeedPrompt( id=self.id, diff --git a/pyrit/models/seeds/seed_simulated_conversation.py b/pyrit/models/seeds/seed_simulated_conversation.py index f5c29525dc..6a9df617a4 100644 --- a/pyrit/models/seeds/seed_simulated_conversation.py +++ b/pyrit/models/seeds/seed_simulated_conversation.py @@ -46,7 +46,7 @@ class SeedSimulatedConversation(Seed): """ Configuration for generating a simulated conversation dynamically. - This class holds the paths and parameters needed to generate prepended conversation + This class holds the prompts, paths, and parameters needed to generate prepended conversation content by running an adversarial chat against a simulated (compliant) target. This is a pure configuration class. The actual generation is performed by @@ -58,7 +58,8 @@ class SeedSimulatedConversation(Seed): Attributes: num_turns: Number of conversation turns to generate. - adversarial_chat_system_prompt_path: Path to the adversarial chat system prompt YAML. + adversarial_chat_system_prompt_path: Legacy path to the adversarial chat system prompt YAML. + adversarial_chat_system_prompt: Canonical inline adversarial chat system prompt. simulated_target_system_prompt_path: Path to the simulated target system prompt YAML. Defaults to the compliant prompt if not specified. next_message_system_prompt_path: Optional path to the system prompt for generating @@ -85,7 +86,8 @@ class SeedSimulatedConversation(Seed): num_turns: int = 3 sequence: int = 0 - adversarial_chat_system_prompt_path: Path + adversarial_chat_system_prompt_path: Path | None = None + adversarial_chat_system_prompt: SeedPrompt | None = None simulated_target_system_prompt_path: Path = SimulatedTargetSystemPromptPaths.COMPLIANT.value next_message_system_prompt_path: Path | None = None pyrit_version: str | None = None @@ -116,6 +118,12 @@ def _default_simulated_target_path(cls, value: Any) -> Any: @model_validator(mode="after") def _validate_and_compute_value(self) -> SeedSimulatedConversation: + has_prompt_path = self.adversarial_chat_system_prompt_path is not None + has_inline_prompt = self.adversarial_chat_system_prompt is not None + if has_prompt_path == has_inline_prompt: + raise ValueError( + "Set exactly one of adversarial_chat_system_prompt_path or adversarial_chat_system_prompt." + ) if self.num_turns <= 0: raise ValueError("num_turns must be a positive integer") if self.sequence < 0: @@ -133,18 +141,63 @@ def _compute_value(self) -> str: str: Deterministic JSON representation of this configuration. """ - config = { + config: dict[str, Any] = { "num_turns": self.num_turns, "sequence": self.sequence, - "adversarial_chat_system_prompt_path": str(self.adversarial_chat_system_prompt_path), "simulated_target_system_prompt_path": str(self.simulated_target_system_prompt_path), "next_message_system_prompt_path": ( str(self.next_message_system_prompt_path) if self.next_message_system_prompt_path else None ), "pyrit_version": self.pyrit_version, } + if self.adversarial_chat_system_prompt is not None: + config["adversarial_chat_system_prompt"] = self._serialize_adversarial_chat_system_prompt() + else: + config["adversarial_chat_system_prompt_path"] = str(self.adversarial_chat_system_prompt_path) return json.dumps(config, sort_keys=True, separators=(",", ":")) + def _serialize_adversarial_chat_system_prompt(self) -> dict[str, Any]: + """ + Serialize the inline prompt without generated identity or timestamp fields. + + Returns: + A deterministic JSON-compatible prompt representation. + """ + assert self.adversarial_chat_system_prompt is not None + prompt_data = self.adversarial_chat_system_prompt.model_dump( + mode="python", + exclude={"id", "date_added", "value_sha256", "prompt_group_id"}, + ) + self._reject_unordered_collections(prompt_data) + return self.adversarial_chat_system_prompt.model_dump( + mode="json", + exclude={"id", "date_added", "value_sha256", "prompt_group_id"}, + ) + + @classmethod + def _reject_unordered_collections(cls, value: Any) -> None: + """ + Reject values whose JSON list order can vary across processes. + + Args: + value: Prompt data to inspect recursively. + + Raises: + ValueError: If the prompt contains a set or frozenset. + """ + if isinstance(value, (set, frozenset)): + raise ValueError( + "Inline adversarial chat system prompts must use ordered JSON-compatible values; " + "set and frozenset values are not supported." + ) + if isinstance(value, dict): + for nested_key, nested_value in value.items(): + cls._reject_unordered_collections(nested_key) + cls._reject_unordered_collections(nested_value) + elif isinstance(value, (list, tuple)): + for nested_value in value: + cls._reject_unordered_collections(nested_value) + def get_identifier(self) -> dict[str, Any]: """ Get an identifier dict capturing this configuration for comparison/storage. @@ -153,17 +206,21 @@ def get_identifier(self) -> dict[str, Any]: Dictionary with configuration details. """ - return { + identifier: dict[str, Any] = { "__type__": "SeedSimulatedConversation", "num_turns": self.num_turns, "sequence": self.sequence, - "adversarial_chat_system_prompt_path": str(self.adversarial_chat_system_prompt_path), "simulated_target_system_prompt_path": str(self.simulated_target_system_prompt_path), "next_message_system_prompt_path": ( str(self.next_message_system_prompt_path) if self.next_message_system_prompt_path else None ), "pyrit_version": self.pyrit_version, } + if self.adversarial_chat_system_prompt is not None: + identifier["adversarial_chat_system_prompt"] = self._serialize_adversarial_chat_system_prompt() + else: + identifier["adversarial_chat_system_prompt_path"] = str(self.adversarial_chat_system_prompt_path) + return identifier def compute_hash(self) -> str: """ @@ -242,6 +299,15 @@ def __repr__(self) -> str: """ has_next_msg = self.next_message_system_prompt_path is not None + if self.adversarial_chat_system_prompt is not None: + adversarial_source = self.adversarial_chat_system_prompt.name or "" + return ( + f"" + ) + + assert self.adversarial_chat_system_prompt_path is not None return ( f" SeedPrompt: + """ + Resolve one simulated-conversation adversarial prompt source to a ``SeedPrompt``. + + The preferred source accepts either an inline prompt or a YAML path. The + ``adversarial_chat_system_prompt_path`` parameter remains as a compatibility + alias. When neither is explicit, ``default_system_prompt_path`` supplies the + factory's conventional name-based YAML fallback. + + Args: + adversarial_chat_system_prompt: Inline prompt or YAML prompt path. + adversarial_chat_system_prompt_path: Legacy YAML prompt path alias. + default_system_prompt_path: YAML fallback used when no explicit source is provided. + + Returns: + The canonical inline prompt. + + Raises: + ValueError: If both sources are provided or no source/default is available. + TypeError: If the preferred source is neither a SeedPrompt nor a Path. + """ + if adversarial_chat_system_prompt_path is not None and adversarial_chat_system_prompt is not None: + raise ValueError("Set only one of adversarial_chat_system_prompt_path or adversarial_chat_system_prompt.") + + if adversarial_chat_system_prompt is not None: + if isinstance(adversarial_chat_system_prompt, SeedPrompt): + return adversarial_chat_system_prompt + if not isinstance(adversarial_chat_system_prompt, Path): + raise TypeError( + "adversarial_chat_system_prompt must be a SeedPrompt or pathlib.Path; " + "use adversarial_chat_system_prompt_path for legacy string paths." + ) + return SeedPrompt.from_yaml_file(adversarial_chat_system_prompt) + + prompt_path = ( + adversarial_chat_system_prompt_path + if adversarial_chat_system_prompt_path is not None + else default_system_prompt_path + ) + if prompt_path is None: + raise ValueError( + "Set one of adversarial_chat_system_prompt or adversarial_chat_system_prompt_path, " + "or provide default_system_prompt_path." + ) + + return SeedPrompt.from_yaml_file(prompt_path) diff --git a/pyrit/scenario/scenarios/airt/psychosocial.py b/pyrit/scenario/scenarios/airt/psychosocial.py index 83df78d6e2..062d792b17 100644 --- a/pyrit/scenario/scenarios/airt/psychosocial.py +++ b/pyrit/scenario/scenarios/airt/psychosocial.py @@ -547,7 +547,7 @@ async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list base_factory = AttackTechniqueFactory.with_simulated_conversation( name=f"psychosocial_{harm.name}", - adversarial_chat_system_prompt_path=harm.escalation_prompt_path, + adversarial_chat_system_prompt=harm.escalation_prompt_path, num_turns=max_turns, ) diff --git a/pyrit/scenario/scenarios/airt/scam.py b/pyrit/scenario/scenarios/airt/scam.py index b7a64dd7be..2ac9727db0 100644 --- a/pyrit/scenario/scenarios/airt/scam.py +++ b/pyrit/scenario/scenarios/airt/scam.py @@ -195,7 +195,7 @@ def _get_atomic_attack_from_technique(self, *, technique: str, seed_groups: list # objective is delivered to the target. role_play_technique = AttackTechniqueFactory.with_simulated_conversation( name="role_play_persuasion_written", - adversarial_chat_system_prompt_path=EXECUTOR_RED_TEAM_PATH + adversarial_chat_system_prompt=EXECUTOR_RED_TEAM_PATH / "role_play" / "role_play_persuasion_written.yaml", next_message_system_prompt_path=EXECUTOR_SIMULATED_TARGET_PATH / "role_play_next_message.yaml", @@ -219,7 +219,7 @@ def _get_atomic_attack_from_technique(self, *, technique: str, seed_groups: list # further, then a fixed "yes." is delivered to the target as the final user turn. context_compliance_technique = AttackTechniqueFactory.with_simulated_conversation( name="context_compliance", - adversarial_chat_system_prompt_path=EXECUTOR_RED_TEAM_PATH + adversarial_chat_system_prompt=EXECUTOR_RED_TEAM_PATH / "context_compliance" / "context_compliance.yaml", simulated_target_system_prompt_path=EXECUTOR_SIMULATED_TARGET_PATH / "context_compliance_target.yaml", diff --git a/pyrit/setup/initializers/techniques/core.py b/pyrit/setup/initializers/techniques/core.py index 8770f8dece..4f73b98285 100644 --- a/pyrit/setup/initializers/techniques/core.py +++ b/pyrit/setup/initializers/techniques/core.py @@ -112,7 +112,7 @@ def get_technique_factories() -> list[AttackTechniqueFactory]: AttackTechniqueFactory.with_simulated_conversation( name="role_play_movie_script", description="Frames the objective as a movie beat by casting the adversarial chat as a screenwriter.", - adversarial_chat_system_prompt_path=( + adversarial_chat_system_prompt=( EXECUTOR_SEED_PROMPT_PATH / "red_teaming" / "role_play" / "role_play_movie_script.yaml" ), next_message_system_prompt_path=EXECUTOR_SIMULATED_TARGET_PATH / "role_play_next_message.yaml", @@ -122,7 +122,7 @@ def get_technique_factories() -> list[AttackTechniqueFactory]: AttackTechniqueFactory.with_simulated_conversation( name="role_play_video_game", description="Frames the objective as an in-game mechanic by casting the adversarial chat as a designer.", - adversarial_chat_system_prompt_path=( + adversarial_chat_system_prompt=( EXECUTOR_SEED_PROMPT_PATH / "red_teaming" / "role_play" / "role_play_video_game.yaml" ), next_message_system_prompt_path=EXECUTOR_SIMULATED_TARGET_PATH / "role_play_next_message.yaml", @@ -132,7 +132,7 @@ def get_technique_factories() -> list[AttackTechniqueFactory]: AttackTechniqueFactory.with_simulated_conversation( name="role_play_trivia_game", description="Frames the objective as a quiz answer by casting the adversarial chat as a trivia host.", - adversarial_chat_system_prompt_path=( + adversarial_chat_system_prompt=( EXECUTOR_SEED_PROMPT_PATH / "red_teaming" / "role_play" / "role_play_trivia_game.yaml" ), next_message_system_prompt_path=EXECUTOR_SIMULATED_TARGET_PATH / "role_play_next_message.yaml", @@ -142,7 +142,7 @@ def get_technique_factories() -> list[AttackTechniqueFactory]: AttackTechniqueFactory.with_simulated_conversation( name="role_play_persuasion", description="Frames a persuasion-resistance test so the target drafts a persuasive spoken script.", - adversarial_chat_system_prompt_path=( + adversarial_chat_system_prompt=( EXECUTOR_SEED_PROMPT_PATH / "red_teaming" / "role_play" / "role_play_persuasion.yaml" ), next_message_system_prompt_path=EXECUTOR_SIMULATED_TARGET_PATH / "role_play_next_message.yaml", @@ -152,7 +152,7 @@ def get_technique_factories() -> list[AttackTechniqueFactory]: AttackTechniqueFactory.with_simulated_conversation( name="role_play_persuasion_written", description="Frames a persuasion-resistance test so the target drafts written persuasive material.", - adversarial_chat_system_prompt_path=( + adversarial_chat_system_prompt=( EXECUTOR_SEED_PROMPT_PATH / "red_teaming" / "role_play" / "role_play_persuasion_written.yaml" ), next_message_system_prompt_path=EXECUTOR_SIMULATED_TARGET_PATH / "role_play_next_message.yaml", @@ -200,9 +200,7 @@ def get_technique_factories() -> list[AttackTechniqueFactory]: AttackTechniqueFactory.with_simulated_conversation( name="context_compliance", description="Injects a fabricated prior exchange so the target continues as if it already agreed.", - adversarial_chat_system_prompt_path=EXECUTOR_RED_TEAM_PATH - / "context_compliance" - / "context_compliance.yaml", + adversarial_chat_system_prompt=EXECUTOR_RED_TEAM_PATH / "context_compliance" / "context_compliance.yaml", simulated_target_system_prompt_path=EXECUTOR_SIMULATED_TARGET_PATH / "context_compliance_target.yaml", final_user_message="yes.", num_turns=1, diff --git a/tests/unit/executor/attack/component/test_simulated_conversation.py b/tests/unit/executor/attack/component/test_simulated_conversation.py index e14909ca40..771160b0b9 100644 --- a/tests/unit/executor/attack/component/test_simulated_conversation.py +++ b/tests/unit/executor/attack/component/test_simulated_conversation.py @@ -12,6 +12,7 @@ from pyrit.executor.attack import AttackConverterConfig, RTASystemPromptPaths from pyrit.executor.attack.multi_turn.simulated_conversation import ( _generate_next_message_async, + _resolve_adversarial_chat_system_prompt_async, generate_simulated_conversation_async, ) from pyrit.models import ( @@ -150,6 +151,97 @@ async def test_raises_error_for_negative_turns( num_turns=-1, ) + async def test_inline_adversarial_prompt_is_forwarded_without_file_loading( + self, + mock_adversarial_chat: MagicMock, + mock_objective_scorer: MagicMock, + sample_conversation: list[Message], + ): + prompt = SeedPrompt( + value="Use {{ objective }}", + parameters=["objective"], + response_json_schema={ + "type": "object", + "properties": {"next_message": {"type": "string"}}, + }, + ) + with ( + patch("pyrit.executor.attack.multi_turn.simulated_conversation.RedTeamingAttack") as mock_attack_class, + patch("pyrit.executor.attack.multi_turn.simulated_conversation.CentralMemory") as mock_memory_class, + ): + mock_attack = MagicMock() + mock_attack.execute_async = AsyncMock( + return_value=AttackResult( + atomic_attack_identifier=ComponentIdentifier( + class_name="RedTeamingAttack", + class_module="pyrit.executor.attack", + ), + conversation_id=str(uuid.uuid4()), + objective="Test objective", + outcome=AttackOutcome.SUCCESS, + executed_turns=3, + ) + ) + mock_attack_class.return_value = mock_attack + mock_memory_class.get_memory_instance.return_value.get_conversation_messages.return_value = iter( + sample_conversation + ) + + await generate_simulated_conversation_async( + objective="Test objective", + adversarial_chat=mock_adversarial_chat, + objective_scorer=mock_objective_scorer, + adversarial_chat_system_prompt=prompt, + ) + + adversarial_config = mock_attack_class.call_args.kwargs["attack_adversarial_config"] + assert adversarial_config.system_prompt is prompt + + async def test_returns_inline_prompt(self): + prompt = SeedPrompt(value="inline") + + resolved = await _resolve_adversarial_chat_system_prompt_async( + adversarial_chat_system_prompt_path=None, + adversarial_chat_system_prompt=prompt, + ) + + assert resolved is prompt + + async def test_loads_legacy_path_off_event_loop(self, tmp_path): + prompt_path = tmp_path / "prompt.yaml" + resolved_prompt = SeedPrompt(value="resolved") + + with patch( + "pyrit.executor.attack.multi_turn.simulated_conversation.asyncio.to_thread", + new_callable=AsyncMock, + return_value=resolved_prompt, + ) as mock_to_thread: + resolved = await _resolve_adversarial_chat_system_prompt_async( + adversarial_chat_system_prompt_path=prompt_path, + adversarial_chat_system_prompt=None, + ) + + assert resolved is resolved_prompt + mock_to_thread.assert_awaited_once_with(SeedPrompt.from_yaml_file, prompt_path) + + @pytest.mark.parametrize( + ("path", "prompt"), + [ + (None, None), + ("prompt.yaml", SeedPrompt(value="inline")), + ], + ) + async def test_rejects_ambiguous_or_missing_adversarial_prompt_source( + self, + path: str | None, + prompt: SeedPrompt | None, + ) -> None: + with pytest.raises(ValueError, match="exactly one"): + await _resolve_adversarial_chat_system_prompt_async( + adversarial_chat_system_prompt_path=path, + adversarial_chat_system_prompt=prompt, + ) + async def test_uses_adversarial_chat_as_simulated_target( self, mock_adversarial_chat: MagicMock, diff --git a/tests/unit/executor/attack/core/test_attack_parameters.py b/tests/unit/executor/attack/core/test_attack_parameters.py index c7bd56811d..82a4a0f3d3 100644 --- a/tests/unit/executor/attack/core/test_attack_parameters.py +++ b/tests/unit/executor/attack/core/test_attack_parameters.py @@ -231,6 +231,38 @@ async def test_generates_simulated_conversation( assert call_kwargs["adversarial_chat"] == mock_adversarial_chat assert call_kwargs["objective_scorer"] == mock_objective_scorer assert call_kwargs["num_turns"] == 3 + config = seed_group_with_simulated_conv.simulated_conversation_config + assert config is not None + assert call_kwargs["adversarial_chat_system_prompt_path"] == config.adversarial_chat_system_prompt_path + assert call_kwargs["adversarial_chat_system_prompt"] is None + + @patch("pyrit.executor.attack.multi_turn.simulated_conversation.generate_simulated_conversation_async") + async def test_forwards_inline_adversarial_prompt( + self, + mock_generate: AsyncMock, + seed_objective: SeedObjective, + mock_adversarial_chat: MagicMock, + mock_objective_scorer: MagicMock, + mock_simulated_result: list, + ) -> None: + prompt = SeedPrompt(value="Use {{ objective }}", parameters=["objective"]) + seed_group = AttackSeedGroup( + seeds=[ + seed_objective, + SeedSimulatedConversation(adversarial_chat_system_prompt=prompt), + ] + ) + mock_generate.return_value = mock_simulated_result + + await AttackParameters.from_seed_group_async( + seed_group=seed_group, + adversarial_chat=mock_adversarial_chat, + objective_scorer=mock_objective_scorer, + ) + + call_kwargs = mock_generate.call_args.kwargs + assert call_kwargs["adversarial_chat_system_prompt"] is prompt + assert call_kwargs["adversarial_chat_system_prompt_path"] is None @patch("pyrit.executor.attack.multi_turn.simulated_conversation.generate_simulated_conversation_async") async def test_uses_generated_prepended_messages( diff --git a/tests/unit/memory/memory_interface/test_interface_seed_prompts.py b/tests/unit/memory/memory_interface/test_interface_seed_prompts.py index 37d256612d..c6d80d3654 100644 --- a/tests/unit/memory/memory_interface/test_interface_seed_prompts.py +++ b/tests/unit/memory/memory_interface/test_interface_seed_prompts.py @@ -11,7 +11,7 @@ from sqlalchemy.exc import SQLAlchemyError from pyrit.memory import MemoryInterface -from pyrit.models import MessagePiece, SeedDataset, SeedGroup, SeedObjective, SeedPrompt +from pyrit.models import MessagePiece, SeedDataset, SeedGroup, SeedObjective, SeedPrompt, SeedSimulatedConversation def assert_original_value_in_list(original_value: str, message_pieces: Sequence[MessagePiece]): @@ -130,6 +130,59 @@ async def test_get_seeds_with_dataset_name_filter(sqlite_instance: MemoryInterfa assert result[0].dataset_name == "dataset1" +async def test_legacy_path_backed_simulated_conversation_persistence_round_trip( + sqlite_instance: MemoryInterface, +) -> None: + seed = SeedSimulatedConversation( + adversarial_chat_system_prompt_path="/legacy/adversarial.yaml", + dataset_name="legacy_simulated", + ) + + await sqlite_instance.add_seeds_to_memory_async(seeds=[seed], added_by="test") + + recovered = sqlite_instance.get_seeds(seed_type="simulated_conversation") + assert len(recovered) == 1 + assert isinstance(recovered[0], SeedSimulatedConversation) + assert recovered[0].adversarial_chat_system_prompt_path == seed.adversarial_chat_system_prompt_path + assert recovered[0].adversarial_chat_system_prompt is None + assert recovered[0].value == seed.value + + +async def test_inline_prompt_simulated_conversation_persistence_round_trip( + sqlite_instance: MemoryInterface, +) -> None: + response_schema = { + "type": "object", + "properties": {"next_message": {"type": "string"}}, + } + seed = SeedSimulatedConversation( + adversarial_chat_system_prompt=SeedPrompt( + value="Use {{ objective }}", + data_type="text", + parameters=["objective"], + response_json_schema=response_schema, + metadata={"source_kind": "inline"}, + is_jinja_template=True, + ), + dataset_name="inline_simulated", + ) + + await sqlite_instance.add_seeds_to_memory_async(seeds=[seed], added_by="test") + + recovered = sqlite_instance.get_seeds(seed_type="simulated_conversation") + assert len(recovered) == 1 + assert isinstance(recovered[0], SeedSimulatedConversation) + assert recovered[0].adversarial_chat_system_prompt_path is None + recovered_prompt = recovered[0].adversarial_chat_system_prompt + assert recovered_prompt is not None + assert recovered_prompt.value == "Use {{ objective }}" + assert recovered_prompt.parameters == ["objective"] + assert recovered_prompt.response_json_schema == response_schema + assert recovered_prompt.metadata == {"source_kind": "inline"} + assert recovered_prompt.is_jinja_template is True + assert recovered[0].value == seed.value + + async def test_get_seeds_with_dataset_name_pattern_startswith(sqlite_instance: MemoryInterface): seed_prompts = [ SeedPrompt(value="prompt1", dataset_name="harm_category_1", data_type="text"), diff --git a/tests/unit/models/test_seed_simulated_conversation.py b/tests/unit/models/test_seed_simulated_conversation.py index c8239a9caa..3df081144c 100644 --- a/tests/unit/models/test_seed_simulated_conversation.py +++ b/tests/unit/models/test_seed_simulated_conversation.py @@ -9,6 +9,7 @@ import pytest from pyrit.models.seeds import ( + SeedPrompt, SeedSimulatedConversation, SimulatedTargetSystemPromptPaths, ) @@ -50,6 +51,25 @@ def test_init_with_minimal_parameters(self, tmp_path): # Default simulated_target_system_prompt_path is the compliant prompt assert conv.simulated_target_system_prompt_path == SimulatedTargetSystemPromptPaths.COMPLIANT.value + def test_init_with_inline_adversarial_prompt(self): + prompt = SeedPrompt(value="Use {{ objective }}", parameters=["objective"], data_type="text") + + conv = SeedSimulatedConversation(adversarial_chat_system_prompt=prompt) + + assert conv.adversarial_chat_system_prompt is prompt + assert conv.adversarial_chat_system_prompt_path is None + + def test_init_rejects_both_adversarial_prompt_sources(self, tmp_path): + with pytest.raises(ValueError, match="exactly one"): + SeedSimulatedConversation( + adversarial_chat_system_prompt_path=tmp_path / "prompt.yaml", + adversarial_chat_system_prompt=SeedPrompt(value="inline"), + ) + + def test_init_rejects_missing_adversarial_prompt_source(self): + with pytest.raises(ValueError, match="exactly one"): + SeedSimulatedConversation() + def test_init_default_num_turns(self, tmp_path): """Test that default num_turns is 3.""" adv_path = tmp_path / "adversarial.yaml" @@ -125,6 +145,57 @@ def test_init_value_is_deterministic(self, tmp_path): assert conv1.value == conv2.value + def test_inline_prompt_value_preserves_template_contract_and_is_deterministic(self): + prompt_kwargs = { + "value": "Use {{ objective }}", + "data_type": "text", + "parameters": ["objective"], + "response_json_schema": { + "type": "object", + "properties": {"next_message": {"type": "string"}}, + }, + "metadata": {"source_kind": "inline"}, + "is_jinja_template": True, + } + + conv1 = SeedSimulatedConversation(adversarial_chat_system_prompt=SeedPrompt(**prompt_kwargs)) + conv2 = SeedSimulatedConversation(adversarial_chat_system_prompt=SeedPrompt(**prompt_kwargs)) + + assert conv1.value == conv2.value + assert conv1.compute_hash() == conv2.compute_hash() + serialized_prompt = json.loads(conv1.value)["adversarial_chat_system_prompt"] + assert serialized_prompt["parameters"] == ["objective"] + assert serialized_prompt["response_json_schema"] == prompt_kwargs["response_json_schema"] + assert serialized_prompt["metadata"] == {"source_kind": "inline"} + assert serialized_prompt["is_jinja_template"] is True + assert "id" not in serialized_prompt + assert "date_added" not in serialized_prompt + + def test_inline_prompt_value_preserves_explicit_null_fields(self): + conv = SeedSimulatedConversation( + adversarial_chat_system_prompt=SeedPrompt( + value="inline", + parameters=None, + metadata=None, + ) + ) + + reconstructed = SeedSimulatedConversation(**json.loads(conv.value)) + prompt = reconstructed.adversarial_chat_system_prompt + assert prompt is not None + assert prompt.parameters is None + assert prompt.metadata is None + assert reconstructed.value == conv.value + + def test_inline_prompt_rejects_unordered_metadata(self): + with pytest.raises(ValueError, match="ordered JSON-compatible values"): + SeedSimulatedConversation( + adversarial_chat_system_prompt=SeedPrompt( + value="inline", + metadata={"tags": {"alpha", "beta"}}, + ) + ) + def test_init_default_sequence_is_zero(self, tmp_path): """Test that default sequence is 0.""" adv_path = tmp_path / "adversarial.yaml" @@ -217,11 +288,11 @@ def test_from_dict_default_num_turns(self, tmp_path): assert conv.num_turns == 3 - def test_from_dict_missing_adversarial_path_raises_error(self): - """Test that construction raises when adversarial path is missing (required field).""" + def test_from_dict_missing_adversarial_source_raises_error(self): + """Test that construction raises when both adversarial prompt sources are missing.""" data = {"num_turns": 3} - with pytest.raises(ValueError, match="adversarial_chat_system_prompt_path"): + with pytest.raises(ValueError, match="exactly one"): SeedSimulatedConversation.model_validate(data) diff --git a/tests/unit/scenario/core/test_attack_technique_factory.py b/tests/unit/scenario/core/test_attack_technique_factory.py index 5ea41cd64b..d13258b0b1 100644 --- a/tests/unit/scenario/core/test_attack_technique_factory.py +++ b/tests/unit/scenario/core/test_attack_technique_factory.py @@ -11,7 +11,13 @@ 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 +from pyrit.models import ( + AttackTechniqueSeedGroup, + ComponentIdentifier, + Identifiable, + SeedPrompt, + SeedSimulatedConversation, +) from pyrit.prompt_normalizer import ConverterConfiguration from pyrit.prompt_target import PromptTarget from pyrit.scenario.core.attack_technique import AttackTechnique @@ -92,6 +98,112 @@ def test_with_simulated_conversation_forwards_description(self): ) assert factory.description == "Staged as a journalist interview." + assert factory.seed_technique is not None + config = factory.seed_technique.simulated_conversation_config + assert isinstance(config, SeedSimulatedConversation) + assert config.adversarial_chat_system_prompt is not None + assert config.adversarial_chat_system_prompt_path is None + + def test_with_simulated_conversation_normalizes_path_to_inline_prompt(self, tmp_path): + prompt_path = tmp_path / "prompt.yaml" + prompt_path.write_text( + "value: Use {{ objective }}\n" + "data_type: text\n" + "parameters:\n" + " - objective\n" + "response_json_schema:\n" + " type: object\n" + " properties:\n" + " next_message:\n" + " type: string\n", + encoding="utf-8", + ) + + factory = AttackTechniqueFactory.with_simulated_conversation( + name="test", + adversarial_chat_system_prompt=prompt_path, + ) + + assert factory.seed_technique is not None + config = factory.seed_technique.simulated_conversation_config + assert isinstance(config, SeedSimulatedConversation) + assert config.adversarial_chat_system_prompt_path is None + assert config.adversarial_chat_system_prompt is not None + assert config.adversarial_chat_system_prompt.parameters == ["objective"] + assert config.adversarial_chat_system_prompt.response_json_schema == { + "type": "object", + "properties": {"next_message": {"type": "string"}}, + } + + def test_with_simulated_conversation_accepts_legacy_path_alias_without_warning(self, tmp_path, recwarn): + prompt_path = tmp_path / "prompt.yaml" + prompt_path.write_text("value: Legacy prompt\ndata_type: text\n", encoding="utf-8") + + factory = AttackTechniqueFactory.with_simulated_conversation( + name="test", + adversarial_chat_system_prompt_path=str(prompt_path), + ) + + assert factory.seed_technique is not None + config = factory.seed_technique.simulated_conversation_config + assert isinstance(config, SeedSimulatedConversation) + assert config.adversarial_chat_system_prompt_path is None + assert config.adversarial_chat_system_prompt is not None + assert config.adversarial_chat_system_prompt.value == "Legacy prompt" + assert not [warning for warning in recwarn if issubclass(warning.category, DeprecationWarning)] + + def test_with_simulated_conversation_rejects_string_on_preferred_prompt_source(self): + with pytest.raises(TypeError, match="pathlib.Path"): + AttackTechniqueFactory.with_simulated_conversation( + name="test", + adversarial_chat_system_prompt="prompt.yaml", # type: ignore[arg-type] + ) + + def test_with_simulated_conversation_accepts_inline_prompt(self): + prompt = SeedPrompt( + value="Use {{ objective }}", + data_type="text", + parameters=["objective"], + metadata={"source_kind": "inline"}, + ) + + factory = AttackTechniqueFactory.with_simulated_conversation( + name="test", + adversarial_chat_system_prompt=prompt, + ) + + assert factory.seed_technique is not None + config = factory.seed_technique.simulated_conversation_config + assert isinstance(config, SeedSimulatedConversation) + assert config.adversarial_chat_system_prompt is prompt + assert config.adversarial_chat_system_prompt_path is None + + def test_with_simulated_conversation_rejects_both_prompt_sources(self, tmp_path): + with pytest.raises(ValueError, match="only one"): + AttackTechniqueFactory.with_simulated_conversation( + name="test", + adversarial_chat_system_prompt_path=tmp_path / "prompt.yaml", + adversarial_chat_system_prompt=SeedPrompt(value="inline"), + ) + + def test_with_simulated_conversation_identifier_is_deterministic_for_inline_prompt(self): + prompt_kwargs = { + "value": "Use {{ objective }}", + "data_type": "text", + "parameters": ["objective"], + "metadata": {"source_kind": "inline"}, + } + + first = AttackTechniqueFactory.with_simulated_conversation( + name="test", + adversarial_chat_system_prompt=SeedPrompt(**prompt_kwargs), + ) + second = AttackTechniqueFactory.with_simulated_conversation( + name="test", + adversarial_chat_system_prompt=SeedPrompt(**prompt_kwargs), + ) + + assert first.get_identifier().hash == second.get_identifier().hash def test_description_does_not_affect_identifier(self): """Description is decorative metadata and must not change the behavioral identity hash.""" diff --git a/tests/unit/scenario/core/test_simulated_conversation_prompt.py b/tests/unit/scenario/core/test_simulated_conversation_prompt.py new file mode 100644 index 0000000000..17987197da --- /dev/null +++ b/tests/unit/scenario/core/test_simulated_conversation_prompt.py @@ -0,0 +1,80 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +"""Tests for simulated-conversation adversarial prompt-source normalization.""" + +from pathlib import Path + +import pytest + +from pyrit.models import SeedPrompt +from pyrit.scenario.core import resolve_simulated_conversation_adversarial_prompt + + +def test_resolve_simulated_conversation_adversarial_prompt_returns_inline_prompt() -> None: + prompt = SeedPrompt(value="Use {{ objective }}", parameters=["objective"], data_type="text") + + resolved = resolve_simulated_conversation_adversarial_prompt( + adversarial_chat_system_prompt=prompt, + ) + + assert resolved is prompt + + +def test_resolve_simulated_conversation_adversarial_prompt_loads_path(tmp_path: Path) -> None: + prompt_path = tmp_path / "prompt.yaml" + prompt_path.write_text( + "value: Use {{ objective }}\ndata_type: text\nparameters:\n - objective\nmetadata:\n source_kind: fixture\n", + encoding="utf-8", + ) + + resolved = resolve_simulated_conversation_adversarial_prompt( + adversarial_chat_system_prompt=prompt_path, + ) + + assert resolved.value == "Use {{ objective }}" + assert resolved.parameters == ["objective"] + assert resolved.metadata == {"source_kind": "fixture"} + assert resolved.is_jinja_template is True + + +def test_resolve_simulated_conversation_adversarial_prompt_accepts_legacy_path_alias(tmp_path: Path) -> None: + prompt_path = tmp_path / "prompt.yaml" + prompt_path.write_text("value: Legacy prompt\ndata_type: text\n", encoding="utf-8") + + resolved = resolve_simulated_conversation_adversarial_prompt( + adversarial_chat_system_prompt_path=str(prompt_path), + ) + + assert resolved.value == "Legacy prompt" + + +def test_resolve_simulated_conversation_adversarial_prompt_rejects_string_on_preferred_source() -> None: + with pytest.raises(TypeError, match="pathlib.Path"): + resolve_simulated_conversation_adversarial_prompt( + adversarial_chat_system_prompt="prompt.yaml", # type: ignore[arg-type] + ) + + +def test_resolve_simulated_conversation_adversarial_prompt_uses_default_path(tmp_path: Path) -> None: + default_path = tmp_path / "default.yaml" + default_path.write_text("value: Default prompt\ndata_type: text\n", encoding="utf-8") + + resolved = resolve_simulated_conversation_adversarial_prompt( + default_system_prompt_path=default_path, + ) + + assert resolved.value == "Default prompt" + + +def test_resolve_simulated_conversation_adversarial_prompt_rejects_both_sources(tmp_path: Path) -> None: + with pytest.raises(ValueError, match="only one"): + resolve_simulated_conversation_adversarial_prompt( + adversarial_chat_system_prompt_path=tmp_path / "prompt.yaml", + adversarial_chat_system_prompt=SeedPrompt(value="inline"), + ) + + +def test_resolve_simulated_conversation_adversarial_prompt_rejects_missing_source() -> None: + with pytest.raises(ValueError, match="Set one of"): + resolve_simulated_conversation_adversarial_prompt() diff --git a/tests/unit/setup/test_technique_initializer.py b/tests/unit/setup/test_technique_initializer.py index b69146ffa8..e4aa0b1b3a 100644 --- a/tests/unit/setup/test_technique_initializer.py +++ b/tests/unit/setup/test_technique_initializer.py @@ -240,11 +240,15 @@ def test_seed_technique_num_turns_matches_canonical_default(self): assert sim is not None assert sim.num_turns == 3 - def test_seed_technique_yaml_path_resolves_to_existing_file(self): + def test_seed_technique_contains_resolved_inline_prompt(self): for f in self._persona_factories(): sim = f.seed_technique.simulated_conversation_config assert sim is not None - assert sim.adversarial_chat_system_prompt_path.exists() + assert sim.adversarial_chat_system_prompt_path is None + prompt = sim.adversarial_chat_system_prompt + assert prompt is not None + assert prompt.name == f.name + assert prompt.parameters == ["objective", "max_turns"] class TestPersonaCrescendoYamls: @@ -323,12 +327,15 @@ def test_final_user_message_is_fixed_affirmation(self): assert yes_prompt.role == "user" assert yes_prompt.sequence == 2 - def test_adversarial_yaml_resolves_to_existing_file(self): + def test_adversarial_prompt_is_resolved_inline(self): factory = self._context_compliance_factory() sim = factory.seed_technique.simulated_conversation_config assert sim is not None - assert sim.adversarial_chat_system_prompt_path.name == "context_compliance.yaml" - assert sim.adversarial_chat_system_prompt_path.exists() + assert sim.adversarial_chat_system_prompt_path is None + prompt = sim.adversarial_chat_system_prompt + assert prompt is not None + assert prompt.name == "context_compliance" + assert prompt.parameters == ["objective", "max_turns"] def test_tagged_core_single_turn_light(self): factory = self._context_compliance_factory() @@ -397,11 +404,15 @@ def test_seed_technique_num_turns_matches_role_play_default(self): assert sim is not None assert sim.num_turns == 2 - def test_seed_technique_yaml_path_resolves_to_existing_file(self): + def test_seed_technique_contains_resolved_inline_prompt(self): for f in self._role_play_factories(): sim = f.seed_technique.simulated_conversation_config assert sim is not None - assert sim.adversarial_chat_system_prompt_path.exists() + assert sim.adversarial_chat_system_prompt_path is None + prompt = sim.adversarial_chat_system_prompt + assert prompt is not None + assert prompt.name == f.name + assert prompt.parameters == ["objective", "max_turns"] def test_all_use_role_play_next_message_prompt(self): for f in self._role_play_factories():