FEAT: Refactor ContextComplianceOrchestrator as ContextComplianceAttack (#1022)

nina-msft · web-flow · commit c3ad49d7c3d5 · 2025-07-25T10:50:03.000-07:00
diff --git a/pyrit/attacks/__init__.py b/pyrit/attacks/__init__.py
@@ -19,6 +19,7 @@
     TAPAttackResult,
     TreeOfAttacksWithPruningAttack,
 )
+from pyrit.attacks.single_turn.context_compliance import ContextComplianceAttack
 from pyrit.attacks.single_turn.flip_attack import FlipAttack
 from pyrit.attacks.single_turn.many_shot_jailbreak import ManyShotJailbreakAttack
 from pyrit.attacks.single_turn.prompt_sending import PromptSendingAttack
@@ -34,6 +35,7 @@
     "AttackStrategy",
     "AttackStrategyLogAdapter",
     "ContextT",
+    "ContextComplianceAttack",
     "ConversationSession",
     "CrescendoAttack",
     "FlipAttack",
diff --git a/pyrit/attacks/single_turn/context_compliance.py b/pyrit/attacks/single_turn/context_compliance.py
@@ -0,0 +1,334 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT license.
+
+import logging
+from pathlib import Path
+from typing import Optional
+
+from pyrit.attacks.base.attack_config import (
+    AttackAdversarialConfig,
+    AttackConverterConfig,
+    AttackScoringConfig,
+)
+from pyrit.attacks.base.attack_context import SingleTurnAttackContext
+from pyrit.attacks.single_turn.prompt_sending import PromptSendingAttack
+from pyrit.common.path import DATASETS_PATH
+from pyrit.models import (
+    PromptRequestPiece,
+    PromptRequestResponse,
+    SeedPrompt,
+    SeedPromptDataset,
+    SeedPromptGroup,
+)
+from pyrit.prompt_normalizer import PromptNormalizer
+from pyrit.prompt_target import PromptChatTarget
+
+logger = logging.getLogger(__name__)
+
+
+class ContextComplianceAttack(PromptSendingAttack):
+    """
+    Implementation of the context compliance attack strategy.
+
+    This attack attempts to bypass safety measures by rephrasing the objective into a more benign context.
+    It uses an adversarial chat target to:
+    1. Rephrase the objective as a more benign question
+    2. Generate a response to the benign question
+    3. Rephrase the original objective as a follow-up question
+
+    This creates a context that makes it harder for the target to detect the true intent.
+    """
+
+    # Default path for context description instructions
+    DEFAULT_CONTEXT_DESCRIPTION_PATH: Path = (
+        Path(DATASETS_PATH) / "orchestrators" / "context_compliance" / "context_description.yaml"
+    )
+
+    # Default affirmative response used in conversation
+    DEFAULT_AFFIRMATIVE_RESPONSE: str = "yes."
+
+    def __init__(
+        self,
+        *,
+        objective_target: PromptChatTarget,
+        attack_adversarial_config: AttackAdversarialConfig,
+        attack_converter_config: Optional[AttackConverterConfig] = None,
+        attack_scoring_config: Optional[AttackScoringConfig] = None,
+        prompt_normalizer: Optional[PromptNormalizer] = None,
+        context_description_instructions_path: Optional[Path] = None,
+        affirmative_response: Optional[str] = None,
+    ) -> None:
+        """
+        Initialize the context compliance attack strategy.
+
+        Args:
+            objective_target (PromptChatTarget): The target system to attack. Must be a PromptChatTarget.
+            attack_adversarial_config (AttackAdversarialConfig): Configuration for the adversarial component,
+                including the adversarial chat target used for rephrasing.
+            attack_converter_config (Optional[AttackConverterConfig]): Configuration for attack converters,
+                including request and response converters.
+            attack_scoring_config (Optional[AttackScoringConfig]): Configuration for attack scoring.
+            prompt_normalizer (Optional[PromptNormalizer]): The prompt normalizer to use for sending prompts.
+            context_description_instructions_path (Optional[Path]): Path to the context description
+                instructions YAML file. If not provided, uses the default path.
+            affirmative_response (Optional[str]): The affirmative response to be used in the conversation history.
+                If not provided, uses the default "yes.".
+
+        Raises:
+            ValueError: If the context description instructions file is invalid.
+        """
+        # Initialize base class
+        super().__init__(
+            objective_target=objective_target,
+            attack_converter_config=attack_converter_config,
+            attack_scoring_config=attack_scoring_config,
+            prompt_normalizer=prompt_normalizer,
+        )
+
+        # Store adversarial chat target
+        self._adversarial_chat = attack_adversarial_config.target
+
+        # Load context description instructions
+        instructions_path = context_description_instructions_path or self.DEFAULT_CONTEXT_DESCRIPTION_PATH
+        self._load_context_description_instructions(instructions_path=instructions_path)
+
+        # Set affirmative response
+        self._affirmative_response = affirmative_response or self.DEFAULT_AFFIRMATIVE_RESPONSE
+
+    def _load_context_description_instructions(self, *, instructions_path: Path) -> None:
+        """
+        Load context description instructions from YAML file.
+
+        Args:
+            instructions_path (Path): Path to the instructions YAML file.
+
+        Raises:
+            ValueError: If the instructions file is invalid or missing required prompts.
+        """
+        try:
+            context_description_instructions = SeedPromptDataset.from_yaml_file(instructions_path)
+        except Exception as e:
+            raise ValueError(f"Failed to load context description instructions from {instructions_path}: {e}")
+
+        if len(context_description_instructions.prompts) < 3:
+            raise ValueError(
+                f"Context description instructions must contain at least 3 prompts, "
+                f"but found {len(context_description_instructions.prompts)}"
+            )
+
+        self._rephrase_objective_to_user_turn = context_description_instructions.prompts[0]
+        self._answer_user_turn = context_description_instructions.prompts[1]
+        self._rephrase_objective_to_question = context_description_instructions.prompts[2]
+
+    def _validate_context(self, *, context: SingleTurnAttackContext) -> None:
+        """
+        Validate the context for the attack.
+        This attack does not support prepended conversations, so it raises an error if one exists.
+        Args:
+            context (SingleTurnAttackContext): The attack context to validate.
+        Raises:
+            ValueError: If the context has a prepended conversation.
+        """
+        # Call parent validation first
+        super()._validate_context(context=context)
+
+        if context.prepended_conversation:
+            raise ValueError(
+                "This attack does not support prepended conversations. "
+                "Please clear the prepended conversation before starting the attack."
+            )
+
+    async def _setup_async(self, *, context: SingleTurnAttackContext) -> None:
+        """
+        Set up the context compliance attack.
+
+        This method:
+        1. Generates a benign rephrasing of the objective
+        2. Gets an answer to the benign question
+        3. Rephrases the objective as a follow-up question
+        4. Constructs a conversation with this context
+        5. Sends an affirmative response to complete the attack
+
+        Args:
+            context (SingleTurnAttackContext): The attack context containing configuration and state.
+        """
+        self._logger.info(f"Setting up context compliance attack for objective: {context.objective}")
+
+        # Build the prepended conversation that creates the benign context
+        prepended_conversation = await self._build_benign_context_conversation_async(
+            objective=context.objective, context=context
+        )
+
+        # Update context with the prepended conversation
+        context.prepended_conversation.extend(prepended_conversation)
+
+        # Create the affirmative seed prompt group
+        affirmative_seed_prompt = SeedPromptGroup(
+            prompts=[
+                SeedPrompt(
+                    value=self._affirmative_response,
+                    data_type="text",
+                )
+            ]
+        )
+
+        # Set the seed prompt group in context
+        context.seed_prompt_group = affirmative_seed_prompt
+
+        await super()._setup_async(context=context)
+
+    async def _build_benign_context_conversation_async(
+        self, *, objective: str, context: SingleTurnAttackContext
+    ) -> list[PromptRequestResponse]:
+        """
+        Build the conversation that creates a benign context for the objective.
+
+        Args:
+            objective (str): The attack objective to disguise.
+            context (SingleTurnAttackContext): The attack context.
+
+        Returns:
+            list[PromptRequestResponse]: The constructed conversation with benign context.
+        """
+        # Step 1: Rephrase objective as a benign question
+        benign_user_query = await self._get_objective_as_benign_question_async(objective=objective, context=context)
+
+        # Step 2: Generate an answer to the benign question
+        benign_answer = await self._get_benign_question_answer_async(
+            benign_user_query=benign_user_query, context=context
+        )
+
+        # Step 3: Rephrase objective as a follow-up question
+        objective_as_question = await self._get_objective_as_question_async(objective=objective, context=context)
+
+        # Construct the assistant response that includes the context
+        assistant_response = self._construct_assistant_response(
+            benign_answer=benign_answer, objective_question=objective_as_question
+        )
+
+        # Create the conversation pieces
+        return [
+            PromptRequestResponse(
+                request_pieces=[
+                    PromptRequestPiece(
+                        role="user",
+                        original_value=objective,
+                        converted_value=benign_user_query,
+                    ),
+                ],
+            ),
+            PromptRequestResponse(
+                request_pieces=[
+                    PromptRequestPiece(
+                        role="assistant",
+                        original_value=assistant_response,
+                    ),
+                ],
+            ),
+        ]
+
+    async def _get_objective_as_benign_question_async(self, *, objective: str, context: SingleTurnAttackContext) -> str:
+        """
+        Rephrase the objective as a more benign question.
+
+        Args:
+            objective (str): The original objective to rephrase.
+            context (SingleTurnAttackContext): The attack context.
+
+        Returns:
+            str: The objective rephrased as a benign question.
+        """
+        seed_prompt_group = SeedPromptGroup(
+            prompts=[
+                SeedPrompt(
+                    value=self._rephrase_objective_to_user_turn.render_template_value(objective=objective),
+                    data_type="text",
+                )
+            ]
+        )
+
+        response = await self._prompt_normalizer.send_prompt_async(
+            seed_prompt_group=seed_prompt_group,
+            target=self._adversarial_chat,
+            orchestrator_identifier=self.get_identifier(),
+            labels=context.memory_labels,
+        )
+
+        return response.get_value()
+
+    async def _get_benign_question_answer_async(
+        self, *, benign_user_query: str, context: SingleTurnAttackContext
+    ) -> str:
+        """
+        Generate an answer to the benign question.
+
+        Args:
+            benign_user_query (str): The benign question to answer.
+            context (SingleTurnAttackContext): The attack context.
+
+        Returns:
+            str: The answer to the benign question.
+        """
+        seed_prompt_group = SeedPromptGroup(
+            prompts=[
+                SeedPrompt(
+                    value=self._answer_user_turn.render_template_value(benign_request=benign_user_query),
+                    data_type="text",
+                )
+            ]
+        )
+
+        response = await self._prompt_normalizer.send_prompt_async(
+            seed_prompt_group=seed_prompt_group,
+            target=self._adversarial_chat,
+            orchestrator_identifier=self.get_identifier(),
+            labels=context.memory_labels,
+        )
+
+        return response.get_value()
+
+    async def _get_objective_as_question_async(self, *, objective: str, context: SingleTurnAttackContext) -> str:
+        """
+        Rephrase the objective as a question.
+
+        Args:
+            objective (str): The original objective to rephrase.
+            context (SingleTurnAttackContext): The attack context.
+
+        Returns:
+            str: The objective rephrased as a question.
+        """
+        seed_prompt_group = SeedPromptGroup(
+            prompts=[
+                SeedPrompt(
+                    value=self._rephrase_objective_to_question.render_template_value(objective=objective),
+                    data_type="text",
+                )
+            ]
+        )
+
+        response = await self._prompt_normalizer.send_prompt_async(
+            seed_prompt_group=seed_prompt_group,
+            target=self._adversarial_chat,
+            orchestrator_identifier=self.get_identifier(),
+            labels=context.memory_labels,
+        )
+
+        return response.get_value()
+
+    def _construct_assistant_response(self, *, benign_answer: str, objective_question: str) -> str:
+        """
+        Construct the assistant response that creates the benign context.
+
+        Args:
+            benign_answer (str): The answer to the benign question.
+            objective_question (str): The objective rephrased as a question.
+
+        Returns:
+            str: The constructed assistant response.
+        """
+        assistant_response = f"{benign_answer}\n\n"
+        assistant_response += "I can provide more information depending on what you're seeking. "
+        assistant_response += f"To help with understanding, {objective_question}"
+
+        return assistant_response
diff --git a/pyrit/attacks/single_turn/prompt_sending.py b/pyrit/attacks/single_turn/prompt_sending.py
@@ -114,8 +114,8 @@ def _validate_context(self, *, context: SingleTurnAttackContext) -> None:
         Raises:
             ValueError: If the context is invalid.
         """
-        if not context.objective:
-            raise ValueError("Attack objective must be provided in the context")
+        if not context.objective or context.objective.isspace():
+            raise ValueError("Attack objective must be provided and non-empty in the context")
 
     async def _setup_async(self, *, context: SingleTurnAttackContext) -> None:
         """
diff --git a/tests/unit/attacks/test_context_compliance.py b/tests/unit/attacks/test_context_compliance.py
