FEAT: Attack Identifier (#1364)

Author: rlundeen2

doc/api.rst

@@ -272,6 +272,7 @@ API Reference
     :toctree: _autosummary/
 
     class_name_to_snake_case
+    AttackIdentifier
     ConverterIdentifier
     Identifiable
     Identifier

pyrit/analytics/result_analysis.py

@@ -62,7 +62,7 @@ def analyze_results(attack_results: list[AttackResult]) -> dict[str, AttackStats
             raise TypeError(f"Expected AttackResult, got {type(attack).__name__}: {attack!r}")
 
         outcome = attack.outcome
-        attack_type = attack.attack_identifier.get("type", "unknown")
+        attack_type = attack.attack_identifier.class_name if attack.attack_identifier else "unknown"
 
         if outcome == AttackOutcome.SUCCESS:
             overall_counts["successes"] += 1

pyrit/exceptions/exception_context.py

@@ -13,9 +13,9 @@
 from contextvars import ContextVar
 from dataclasses import dataclass, field
 from enum import Enum
-from typing import Any, Dict, Optional, Union
+from typing import Any, Optional
 
-from pyrit.identifiers import Identifier
+from pyrit.identifiers import AttackIdentifier, Identifier
 
 
 class ComponentRole(Enum):
@@ -61,11 +61,11 @@ class ExecutionContext:
     # The attack strategy class name (e.g., "PromptSendingAttack")
     attack_strategy_name: Optional[str] = None
 
-    # The identifier from the attack strategy's get_identifier()
-    attack_identifier: Optional[Dict[str, Any]] = None
+    # The identifier for the attack strategy
+    attack_identifier: Optional[AttackIdentifier] = None
 
     # The identifier from the component's get_identifier() (target, scorer, etc.)
-    component_identifier: Optional[Dict[str, Any]] = None
+    component_identifier: Optional[Identifier] = None
 
     # The objective target conversation ID if available
     objective_target_conversation_id: Optional[str] = None
@@ -192,8 +192,8 @@ def execution_context(
     *,
     component_role: ComponentRole,
     attack_strategy_name: Optional[str] = None,
-    attack_identifier: Optional[Dict[str, Any]] = None,
-    component_identifier: Optional[Union[Identifier, Dict[str, Any]]] = None,
+    attack_identifier: Optional[AttackIdentifier] = None,
+    component_identifier: Optional[Identifier] = None,
     objective_target_conversation_id: Optional[str] = None,
     objective: Optional[str] = None,
 ) -> ExecutionContextManager:
@@ -203,9 +203,8 @@ def execution_context(
     Args:
         component_role: The role of the component being executed.
         attack_strategy_name: The name of the attack strategy class.
-        attack_identifier: The identifier from attack.get_identifier().
+        attack_identifier: The attack identifier.
         component_identifier: The identifier from component.get_identifier().
-            Can be an Identifier object or a dict (legacy format).
         objective_target_conversation_id: The objective target conversation ID if available.
         objective: The attack objective if available.
 
@@ -215,22 +214,15 @@ def execution_context(
     # Extract endpoint and component_name from component_identifier if available
     endpoint = None
     component_name = None
-    component_id_dict: Optional[Dict[str, Any]] = None
     if component_identifier:
-        if isinstance(component_identifier, Identifier):
-            endpoint = getattr(component_identifier, "endpoint", None)
-            component_name = component_identifier.class_name
-            component_id_dict = component_identifier.to_dict()
-        else:
-            endpoint = component_identifier.get("endpoint")
-            component_name = component_identifier.get("__type__")
-            component_id_dict = component_identifier
+        endpoint = getattr(component_identifier, "endpoint", None)
+        component_name = component_identifier.class_name
 
     context = ExecutionContext(
         component_role=component_role,
         attack_strategy_name=attack_strategy_name,
         attack_identifier=attack_identifier,
-        component_identifier=component_id_dict,
+        component_identifier=component_identifier,
         objective_target_conversation_id=objective_target_conversation_id,
         endpoint=endpoint,
         component_name=component_name,

pyrit/executor/attack/component/conversation_manager.py

@@ -4,13 +4,13 @@
 import logging
 import uuid
 from dataclasses import dataclass, field
-from typing import TYPE_CHECKING, Any, Dict, List, Optional, Sequence, Union
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Sequence
 
 from pyrit.common.utils import combine_dict
 from pyrit.executor.attack.component.prepended_conversation_config import (
     PrependedConversationConfig,
 )
-from pyrit.identifiers import TargetIdentifier
+from pyrit.identifiers import AttackIdentifier, TargetIdentifier
 from pyrit.memory import CentralMemory
 from pyrit.message_normalizer import ConversationContextNormalizer
 from pyrit.models import ChatMessageRole, Message, MessagePiece, Score
@@ -54,8 +54,8 @@ def get_adversarial_chat_messages(
     prepended_conversation: List[Message],
     *,
     adversarial_chat_conversation_id: str,
-    attack_identifier: Dict[str, str],
-    adversarial_chat_target_identifier: Union[TargetIdentifier, Dict[str, Any]],
+    attack_identifier: AttackIdentifier,
+    adversarial_chat_target_identifier: TargetIdentifier,
     labels: Optional[Dict[str, str]] = None,
 ) -> List[Message]:
     """
@@ -183,7 +183,7 @@ class ConversationManager:
     def __init__(
         self,
         *,
-        attack_identifier: Dict[str, str],
+        attack_identifier: AttackIdentifier,
         prompt_normalizer: Optional[PromptNormalizer] = None,
     ):
         """

pyrit/executor/attack/core/attack_strategy.py

@@ -20,6 +20,7 @@
     StrategyEventData,
     StrategyEventHandler,
 )
+from pyrit.identifiers import AttackIdentifier, Identifiable
 from pyrit.memory.central_memory import CentralMemory
 from pyrit.models import (
     AttackOutcome,
@@ -224,7 +225,7 @@ def _log_attack_outcome(self, result: AttackResult) -> None:
         self._logger.info(message)
 
 
-class AttackStrategy(Strategy[AttackStrategyContextT, AttackStrategyResultT], ABC):
+class AttackStrategy(Strategy[AttackStrategyContextT, AttackStrategyResultT], Identifiable[AttackIdentifier], ABC):
     """
     Abstract base class for attack strategies.
     Defines the interface for executing attacks and handling results.
@@ -258,6 +259,54 @@ def __init__(
         )
         self._objective_target = objective_target
         self._params_type = params_type
+        # Guard so subclasses that set converters before calling super() aren't clobbered
+        if not hasattr(self, "_request_converters"):
+            self._request_converters: list[Any] = []
+        if not hasattr(self, "_response_converters"):
+            self._response_converters: list[Any] = []
+
+    def _build_identifier(self) -> AttackIdentifier:
+        """
+        Build the typed identifier for this attack strategy.
+
+        Captures the objective target, optional scorer, and converter pipeline.
+        This is the *stable* strategy-level identifier that does not change
+        between calls to ``execute_async``.
+
+        Returns:
+            AttackIdentifier: The constructed identifier.
+        """
+        # Get target identifier
+        objective_target_identifier = self.get_objective_target().get_identifier()
+
+        # Get scorer identifier if present
+        scorer_identifier = None
+        scoring_config = self.get_attack_scoring_config()
+        if scoring_config and scoring_config.objective_scorer:
+            scorer_identifier = scoring_config.objective_scorer.get_identifier()
+
+        # Get request converter identifiers if present
+        request_converter_ids = None
+        if self._request_converters:
+            request_converter_ids = [
+                converter.get_identifier() for config in self._request_converters for converter in config.converters
+            ]
+
+        # Get response converter identifiers if present
+        response_converter_ids = None
+        if self._response_converters:
+            response_converter_ids = [
+                converter.get_identifier() for config in self._response_converters for converter in config.converters
+            ]
+
+        return AttackIdentifier(
+            class_name=self.__class__.__name__,
+            class_module=self.__class__.__module__,
+            objective_target_identifier=objective_target_identifier,
+            objective_scorer_identifier=scorer_identifier,
+            request_converter_identifiers=request_converter_ids or None,
+            response_converter_identifiers=response_converter_ids or None,
+        )
 
     @property
     def params_type(self) -> Type[AttackParameters]:
@@ -291,6 +340,15 @@ def get_attack_scoring_config(self) -> Optional[AttackScoringConfig]:
         """
         return None
 
+    def get_request_converters(self) -> list[Any]:
+        """
+        Get request converter configurations used by this strategy.
+
+        Returns:
+            list[Any]: The list of request PromptConverterConfiguration objects.
+        """
+        return self._request_converters
+
     @overload
     async def execute_async(
         self,

pyrit/executor/attack/multi_turn/tree_of_attacks.py

@@ -37,6 +37,7 @@
 )
 from pyrit.executor.attack.core.attack_strategy import AttackStrategy
 from pyrit.executor.attack.multi_turn import MultiTurnAttackContext
+from pyrit.identifiers import AttackIdentifier
 from pyrit.memory import CentralMemory
 from pyrit.models import (
     AttackOutcome,
@@ -267,7 +268,7 @@ def __init__(
         request_converters: List[PromptConverterConfiguration],
         response_converters: List[PromptConverterConfiguration],
         auxiliary_scorers: Optional[List[Scorer]],
-        attack_id: dict[str, str],
+        attack_id: AttackIdentifier,
         attack_strategy_name: str,
         memory_labels: Optional[dict[str, str]] = None,
         parent_id: Optional[str] = None,
@@ -289,7 +290,7 @@ def __init__(
             request_converters (List[PromptConverterConfiguration]): Converters for request normalization
             response_converters (List[PromptConverterConfiguration]): Converters for response normalization
             auxiliary_scorers (Optional[List[Scorer]]): Additional scorers for the response
-            attack_id (dict[str, str]): Unique identifier for the attack.
+            attack_id (AttackIdentifier): Unique identifier for the attack.
             attack_strategy_name (str): Name of the attack strategy for execution context.
             memory_labels (Optional[dict[str, str]]): Labels for memory storage.
             parent_id (Optional[str]): ID of the parent node, if this is a child node

pyrit/executor/attack/printer/console_printer.py

@@ -258,10 +258,8 @@ async def print_summary_async(self, result: AttackResult) -> None:
 
         # Extract attack type name from attack_identifier
         attack_type = "Unknown"
-        if isinstance(result.attack_identifier, dict) and "__type__" in result.attack_identifier:
-            attack_type = result.attack_identifier["__type__"]
-        elif isinstance(result.attack_identifier, str):
-            attack_type = result.attack_identifier
+        if result.attack_identifier:
+            attack_type = result.attack_identifier.class_name
 
         self._print_colored(f"{self._indent * 2}• Attack Type: {attack_type}", Fore.CYAN)
         self._print_colored(f"{self._indent * 2}• Conversation ID: {result.conversation_id}", Fore.CYAN)

pyrit/executor/attack/printer/markdown_printer.py

@@ -493,7 +493,7 @@ async def _get_summary_markdown_async(self, result: AttackResult) -> List[str]:
         markdown_lines.append("|-------|-------|")
         markdown_lines.append(f"| **Objective** | {result.objective} |")
 
-        attack_type = result.attack_identifier.get("__type__", "Unknown")
+        attack_type = result.attack_identifier.class_name if result.attack_identifier else "Unknown"
 
         markdown_lines.append(f"| **Attack Type** | `{attack_type}` |")
         markdown_lines.append(f"| **Conversation ID** | `{result.conversation_id}` |")

pyrit/executor/benchmark/fairness_bias.py

@@ -17,6 +17,7 @@
     PromptSendingAttack,
 )
 from pyrit.executor.core import Strategy, StrategyContext
+from pyrit.identifiers import AttackIdentifier
 from pyrit.memory import CentralMemory
 from pyrit.models import (
     AttackOutcome,
@@ -195,7 +196,10 @@ async def _perform_async(self, *, context: FairnessBiasBenchmarkContext) -> Atta
                 conversation_id=str(uuid.UUID(int=0)),
                 objective=context.generated_objective,
                 outcome=AttackOutcome.FAILURE,
-                attack_identifier=self.get_identifier(),
+                attack_identifier=AttackIdentifier(
+                    class_name=self.__class__.__name__,
+                    class_module=self.__class__.__module__,
+                ),
             )
 
         return last_attack_result

pyrit/executor/core/strategy.py

@@ -176,19 +176,6 @@ def __init__(
             default_values.get_non_required_value(env_var_name="GLOBAL_MEMORY_LABELS") or "{}"
         )
 
-    def get_identifier(self) -> Dict[str, str]:
-        """
-        Get a serializable identifier for the strategy instance.
-
-        Returns:
-            dict: A dictionary containing the type, module, and unique ID of the strategy.
-        """
-        return {
-            "__type__": self.__class__.__name__,
-            "__module__": self.__class__.__module__,
-            "id": str(self._id),
-        }
-
     def _register_event_handler(self, event_handler: StrategyEventHandler[StrategyContextT, StrategyResultT]) -> None:
         """
         Register an event handler for strategy events.