FEAT: atomic attack identifier (microsoft#1446)

Author: rlundeen2

doc/api.rst

@@ -271,12 +271,17 @@ API Reference
     :nosignatures:
     :toctree: _autosummary/
 
+    AtomicAttackEvaluationIdentifier
+    build_atomic_attack_identifier
+    build_seed_identifier
+    ChildEvalRule
     class_name_to_snake_case
     ComponentIdentifier
     compute_eval_hash
     config_hash
-    EvaluationIdentity
+    EvaluationIdentifier
     Identifiable
+    ScorerEvaluationIdentifier
     snake_case_to_class_name
 
 :py:mod:`pyrit.memory`

doc/code/memory/3_memory_data_types.md

@@ -151,6 +151,7 @@ Scores enable automated evaluation of attack success, content harmfulness, and o
 - **`conversation_id`**: The conversation that produced this result
 - **`objective`**: Natural-language description of the attacker's goal
 - **`attack_identifier`**: Information identifying the attack strategy used
+- **`atomic_attack_identifier`**: Composite identifier combining the attack strategy with general technique seed identifiers from the dataset
 - **`last_response`**: The final `MessagePiece` generated in the attack
 - **`last_score`**: The final score assigned to the last response
 - **`executed_turns`**: Number of turns executed in the attack

pyrit/analytics/result_analysis.py

@@ -62,7 +62,9 @@ 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.class_name if attack.attack_identifier else "unknown"
+        attack_type = (
+            attack.get_attack_strategy_identifier().class_name if attack.get_attack_strategy_identifier() else "unknown"
+        )
 
         if outcome == AttackOutcome.SUCCESS:
             overall_counts["successes"] += 1

pyrit/backend/mappers/attack_mappers.py

@@ -204,7 +204,7 @@ def attack_result_to_summary(
     created_at = datetime.fromisoformat(created_str) if created_str else datetime.now(timezone.utc)
     updated_at = datetime.fromisoformat(updated_str) if updated_str else created_at
 
-    aid = ar.attack_identifier
+    aid = ar.get_attack_strategy_identifier()
 
     # Extract only frontend-relevant fields from ComponentIdentifier
     target_id = aid.get_child("objective_target") if aid else None

pyrit/backend/services/attack_service.py

@@ -50,6 +50,7 @@
 from pyrit.backend.services.converter_service import get_converter_service
 from pyrit.backend.services.target_service import get_target_service
 from pyrit.identifiers import ComponentIdentifier
+from pyrit.identifiers.atomic_attack_identifier import build_atomic_attack_identifier
 from pyrit.memory import CentralMemory
 from pyrit.models import (
     AttackOutcome,
@@ -295,10 +296,12 @@ async def create_attack_async(self, *, request: CreateAttackRequest) -> CreateAt
         attack_result = AttackResult(
             conversation_id=conversation_id,
             objective=request.name or "Manual attack via GUI",
-            attack_identifier=ComponentIdentifier(
-                class_name=request.name or "ManualAttack",
-                class_module="pyrit.backend",
-                children={"objective_target": target_identifier} if target_identifier else {},
+            atomic_attack_identifier=build_atomic_attack_identifier(
+                attack_identifier=ComponentIdentifier(
+                    class_name=request.name or "ManualAttack",
+                    class_module="pyrit.backend",
+                    children={"objective_target": target_identifier} if target_identifier else {},
+                ),
             ),
             outcome=AttackOutcome.UNDETERMINED,
             metadata={

pyrit/executor/attack/core/attack_executor.py

@@ -9,7 +9,7 @@
 
 import asyncio
 from collections.abc import Iterator, Sequence
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -56,6 +56,13 @@ class AttackExecutorResult(Generic[AttackResultT]):
 
     completed_results: list[AttackResultT]
     incomplete_objectives: list[tuple[str, BaseException]]
+    input_indices: list[int] = field(default_factory=list)
+    """Maps each completed result to its position in the original input sequence.
+
+    ``input_indices[i]`` is the index in the original objectives/seed_groups/params
+    list that produced ``completed_results[i]``.  When some inputs fail, this lets
+    callers correlate results back to the specific input that produced them.
+    """
 
     def __iter__(self) -> Iterator[AttackResultT]:
         """
@@ -331,16 +338,19 @@ def _process_execution_results(
         """
         completed: list[AttackStrategyResultT] = []
         incomplete: list[tuple[str, BaseException]] = []
+        completed_indices: list[int] = []
 
-        for objective, result in zip(objectives, results_or_exceptions, strict=False):
+        for i, (objective, result) in enumerate(zip(objectives, results_or_exceptions, strict=False)):
             if isinstance(result, BaseException):
                 incomplete.append((objective, result))
             else:
                 completed.append(result)
+                completed_indices.append(i)
 
         executor_result: AttackExecutorResult[AttackStrategyResultT] = AttackExecutorResult(
             completed_results=completed,
             incomplete_objectives=incomplete,
+            input_indices=completed_indices,
         )
 
         if not return_partial_on_failure:

pyrit/executor/attack/multi_turn/chunked_request.py

@@ -20,6 +20,7 @@
     MultiTurnAttackContext,
     MultiTurnAttackStrategy,
 )
+from pyrit.identifiers import build_atomic_attack_identifier
 from pyrit.models import (
     AttackOutcome,
     AttackResult,
@@ -316,7 +317,7 @@ async def _perform_async(self, *, context: ChunkedRequestAttackContext) -> Attac
         return AttackResult(
             conversation_id=context.session.conversation_id,
             objective=context.objective,
-            attack_identifier=self.get_identifier(),
+            atomic_attack_identifier=build_atomic_attack_identifier(attack_identifier=self.get_identifier()),
             last_response=response.get_piece() if response else None,
             last_score=score,
             related_conversations=context.related_conversations,

pyrit/executor/attack/multi_turn/crescendo.py

@@ -30,6 +30,7 @@
     MultiTurnAttackContext,
     MultiTurnAttackStrategy,
 )
+from pyrit.identifiers import build_atomic_attack_identifier
 from pyrit.memory.central_memory import CentralMemory
 from pyrit.message_normalizer import ConversationContextNormalizer
 from pyrit.models import (
@@ -393,7 +394,7 @@ async def _perform_async(self, *, context: CrescendoAttackContext) -> CrescendoA
 
         # Prepare the result
         result = CrescendoAttackResult(
-            attack_identifier=self.get_identifier(),
+            atomic_attack_identifier=build_atomic_attack_identifier(attack_identifier=self.get_identifier()),
             conversation_id=context.session.conversation_id,
             objective=context.objective,
             outcome=(AttackOutcome.SUCCESS if achieved_objective else AttackOutcome.FAILURE),

pyrit/executor/attack/multi_turn/multi_prompt_sending.py

@@ -19,6 +19,7 @@
     MultiTurnAttackContext,
     MultiTurnAttackStrategy,
 )
+from pyrit.identifiers import build_atomic_attack_identifier
 from pyrit.models import (
     AttackOutcome,
     AttackResult,
@@ -287,7 +288,7 @@ async def _perform_async(self, *, context: MultiTurnAttackContext[Any]) -> Attac
         return AttackResult(
             conversation_id=context.session.conversation_id,
             objective=context.objective,
-            attack_identifier=self.get_identifier(),
+            atomic_attack_identifier=build_atomic_attack_identifier(attack_identifier=self.get_identifier()),
             last_response=response.get_piece() if response else None,
             last_score=score,
             related_conversations=context.related_conversations,

pyrit/executor/attack/multi_turn/red_teaming.py

@@ -26,6 +26,7 @@
     MultiTurnAttackContext,
     MultiTurnAttackStrategy,
 )
+from pyrit.identifiers import build_atomic_attack_identifier
 from pyrit.memory import CentralMemory
 from pyrit.models import (
     AttackOutcome,
@@ -314,7 +315,7 @@ async def _perform_async(self, *, context: MultiTurnAttackContext[Any]) -> Attac
 
         # Prepare the result
         return AttackResult(
-            attack_identifier=self.get_identifier(),
+            atomic_attack_identifier=build_atomic_attack_identifier(attack_identifier=self.get_identifier()),
             conversation_id=context.session.conversation_id,
             objective=context.objective,
             outcome=(AttackOutcome.SUCCESS if achieved_objective else AttackOutcome.FAILURE),