BUG: Stop leaking absolute media paths and SAS tokens in Attack History 'Last Message'

pyrit/backend/mappers/__init__.py

@@ -8,6 +8,7 @@
 Centralizes all translation logic so domain models can evolve independently of the API contract.
 """
 
+from pyrit.backend.mappers._preview import format_last_message_preview
 from pyrit.backend.mappers.attack_mappers import (
     attack_result_to_summary,
     pyrit_messages_to_dto_async,
@@ -25,6 +26,7 @@
 __all__ = [
     "attack_result_to_summary",
     "converter_object_to_instance",
+    "format_last_message_preview",
     "pyrit_messages_to_dto_async",
     "pyrit_scores_to_dto",
     "request_piece_to_pyrit_message_piece",

pyrit/backend/mappers/_preview.py

@@ -0,0 +1,95 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT license.
+
+"""
+Presentation-layer formatter for ``ConversationStats.last_message_preview``.
+
+Lives in the backend mapper package because the formatting it produces
+(``[Image: <basename>]`` etc.) is purely a display concern for the GUI API
+responses — the memory layer stays data-agnostic and just stores the raw
+value + data type.
+
+The motivating bug: ``converted_value`` for media-path data types
+(``image_path`` / ``audio_path`` / ``video_path`` / ``binary_path``) is a
+filesystem path or blob URL. Rendering it raw in the Attack History preview
+leaks the absolute on-disk location of memory artifacts
+(e.g. ``C:\\Users\\<name>\\git\\PyRIT\\dbdata\\...\\1780.mp3``).
+"""
+
+from pathlib import PureWindowsPath
+from urllib.parse import urlparse
+
+from pyrit.models import MEDIA_PATH_DATA_TYPES, ConversationStats
+
+# Friendly label per media-path data type. Kept here next to the formatter
+# so adding a new media type only requires updating one place.
+_MEDIA_LABEL: dict[str, str] = {
+    "image_path": "Image",
+    "audio_path": "Audio",
+    "video_path": "Video",
+    "binary_path": "File",
+}
+
+
+def _derive_basename(value: str) -> str | None:
+    """
+    Return a display-safe basename for *value*.
+
+    Args:
+        value: A filesystem path, URL, or other reference.
+
+    Returns:
+        The basename (filename portion) of *value*, or ``None`` if one can't
+        be derived (e.g. data URI, empty value).
+    """
+    if not value or value.startswith("data:"):
+        return None
+    if value.startswith(("http://", "https://")):
+        # Strip query string (e.g. SAS tokens) before taking the basename.
+        parsed = urlparse(value)
+        name = PureWindowsPath(parsed.path).name
+        return name or None
+    # Local path — PureWindowsPath treats both ``/`` and ``\`` as separators,
+    # so Windows-style paths stored from a Windows host are split correctly
+    # even when this code runs on a POSIX host (CI, Linux deployments).
+    return PureWindowsPath(value).name or None
+
+
+def format_last_message_preview(
+    *,
+    value: str | None,
+    data_type: str | None,
+    max_len: int = ConversationStats.PREVIEW_MAX_LEN,
+) -> str | None:
+    """
+    Build a display string for ``ConversationStats.last_message_preview``.
+
+    Media-path data types are rendered as ``[Image: <basename>]`` (and
+    variants) so the absolute filesystem path of memory artifacts is never
+    exposed through API responses or UI previews. Text-like data types pass
+    through with truncation and an ellipsis suffix when they exceed
+    *max_len*.
+
+    Args:
+        value: Raw ``converted_value`` for the last piece (or ``None``).
+        data_type: ``converted_value_data_type`` for that piece. ``None``
+            falls back to the text path.
+        max_len: Maximum length for text previews before truncation.
+
+    Returns:
+        The formatted preview string, or ``None`` when there is nothing
+        meaningful to show.
+    """
+    if data_type in MEDIA_PATH_DATA_TYPES:
+        # MEDIA_PATH_DATA_TYPES guarantees ``data_type`` is a key in
+        # ``_MEDIA_LABEL`` — both are derived from the same source list.
+        label = _MEDIA_LABEL[data_type]
+        basename = _derive_basename(value or "")
+        return f"[{label}: {basename}]" if basename else f"[{label}]"
+
+    if not value:
+        return None
+
+    if len(value) > max_len:
+        return value[:max_len] + "..."
+    return value

pyrit/backend/mappers/attack_mappers.py

@@ -24,6 +24,7 @@
 from azure.storage.blob import ContainerSasPermissions, generate_container_sas
 from azure.storage.blob.aio import BlobServiceClient
 
+from pyrit.backend.mappers._preview import format_last_message_preview
 from pyrit.backend.models.attacks import (
     AddMessageRequest,
     AttackSummary,
@@ -35,7 +36,7 @@
     TargetInfo,
 )
 from pyrit.common.deprecation import print_deprecation_message
-from pyrit.models import AttackResult, ChatMessageRole, PromptDataType
+from pyrit.models import MEDIA_PATH_DATA_TYPES, AttackResult, ChatMessageRole, PromptDataType
 from pyrit.models import Message as PyritMessage
 from pyrit.models import MessagePiece as PyritMessagePiece
 from pyrit.models import Score as PyritScore
@@ -50,9 +51,6 @@
 # Domain → DTO  (for API responses)
 # ============================================================================
 
-# Media data types whose values are file paths (local or Azure Blob URLs)
-_MEDIA_PATH_TYPES = frozenset({"image_path", "audio_path", "video_path", "binary_path"})
-
 # ---------------------------------------------------------------------------
 # Azure Blob SAS token cache
 # ---------------------------------------------------------------------------
@@ -172,7 +170,7 @@ def _resolve_media_url(*, value: Optional[str], data_type: str) -> Optional[str]
         The value unchanged for non-media types, a ``/api/media?path=...``
         URL for local file paths, or the original value for blob URLs / data URIs.
     """
-    if not value or data_type not in _MEDIA_PATH_TYPES:
+    if not value or data_type not in MEDIA_PATH_DATA_TYPES:
         return value
     # Already a URL or data URI — pass through
     if value.startswith(("http://", "https://", "data:")):
@@ -227,7 +225,10 @@ def attack_result_to_summary(
         AttackSummary DTO ready for the API response.
     """
     message_count = stats.message_count
-    last_preview = stats.last_message_preview
+    last_preview = format_last_message_preview(
+        value=stats.last_message_preview,
+        data_type=stats.last_message_data_type,
+    )
 
     # Merge attack-result labels with conversation-level labels.
     # Conversation labels take precedence on key collision.
@@ -297,7 +298,9 @@ def pyrit_scores_to_dto(scores: list[PyritScore]) -> list[Score]:
     return [
         Score(
             score_id=str(score.id),
-            scorer_type=score.scorer_class_identifier.class_name,
+            scorer_type=(
+                score.scorer_class_identifier.class_name or "Unknown" if score.scorer_class_identifier else "Unknown"
+            ),
             score_type=score.score_type,
             score_value=score.score_value,
             score_category=score.score_category,

pyrit/backend/services/attack_service.py

@@ -24,8 +24,9 @@
 from typing import Any, Literal, cast
 from urllib.parse import parse_qs, urlparse
 
-from pyrit.backend.mappers.attack_mappers import (
+from pyrit.backend.mappers import (
     attack_result_to_summary,
+    format_last_message_preview,
     pyrit_messages_to_dto_async,
     request_piece_to_pyrit_message_piece,
     request_to_pyrit_message,
@@ -177,11 +178,13 @@ async def list_attacks_async(
 
             total_count = (main_stats.message_count if main_stats else 0) + sum(s.message_count for s in pruned_stats)
             preview = main_stats.last_message_preview if main_stats else None
+            preview_data_type = main_stats.last_message_data_type if main_stats else None
             conv_labels = (main_stats.labels if main_stats else None) or {}
 
             merged = ConversationStats(
                 message_count=total_count,
                 last_message_preview=preview,
+                last_message_data_type=preview_data_type,
                 labels=conv_labels,
             )
 
@@ -419,7 +422,10 @@ async def get_conversations_async(self, *, attack_result_id: str) -> AttackConve
                 ConversationSummary(
                     conversation_id=conv_id,
                     message_count=stats.message_count if stats else 0,
-                    last_message_preview=stats.last_message_preview if stats else None,
+                    last_message_preview=format_last_message_preview(
+                        value=stats.last_message_preview if stats else None,
+                        data_type=stats.last_message_data_type if stats else None,
+                    ),
                     created_at=created_at,
                 )
             )

pyrit/memory/azure_sql_memory.py

@@ -615,18 +615,23 @@ def get_conversation_stats(self, *, conversation_ids: Sequence[str]) -> dict[str
         placeholders = ", ".join(f":cid{i}" for i in range(len(conversation_ids)))
         params = {f"cid{i}": cid for i, cid in enumerate(conversation_ids)}
 
-        max_len = ConversationStats.PREVIEW_MAX_LEN
         sql = text(
             f"""
             SELECT
                 pme.conversation_id,
                 COUNT(DISTINCT pme.sequence) AS msg_count,
                 (
-                    SELECT TOP 1 LEFT(p2.converted_value, {max_len + 3})
+                    SELECT TOP 1 LEFT(p2.converted_value, {ConversationStats.PREVIEW_FETCH_MAX_LEN})
                     FROM "PromptMemoryEntries" p2
                     WHERE p2.conversation_id = pme.conversation_id
                     ORDER BY p2.sequence DESC, p2.id DESC
                 ) AS last_preview,
+                (
+                    SELECT TOP 1 p2b.converted_value_data_type
+                    FROM "PromptMemoryEntries" p2b
+                    WHERE p2b.conversation_id = pme.conversation_id
+                    ORDER BY p2b.sequence DESC, p2b.id DESC
+                ) AS last_data_type,
                 (
                     SELECT TOP 1 p3.labels
                     FROM "PromptMemoryEntries" p3
@@ -648,11 +653,7 @@ def get_conversation_stats(self, *, conversation_ids: Sequence[str]) -> dict[str
 
         result: dict[str, ConversationStats] = {}
         for row in rows:
-            conv_id, msg_count, last_preview, raw_labels, raw_created_at = row
-
-            preview = None
-            if last_preview:
-                preview = last_preview[:max_len] + "..." if len(last_preview) > max_len else last_preview
+            conv_id, msg_count, last_preview, last_data_type, raw_labels, raw_created_at = row
 
             labels: dict[str, str] = {}
             if raw_labels and raw_labels not in ("null", "{}"):
@@ -668,7 +669,8 @@ def get_conversation_stats(self, *, conversation_ids: Sequence[str]) -> dict[str
 
             result[conv_id] = ConversationStats(
                 message_count=msg_count,
-                last_message_preview=preview,
+                last_message_preview=last_preview,
+                last_message_data_type=last_data_type,
                 labels=labels,
                 created_at=created_at,
             )

pyrit/memory/sqlite_memory.py

@@ -739,19 +739,25 @@ def get_conversation_stats(self, *, conversation_ids: Sequence[str]) -> dict[str
         placeholders = ", ".join(f":cid{i}" for i in range(len(conversation_ids)))
         params = {f"cid{i}": cid for i, cid in enumerate(conversation_ids)}
 
-        max_len = ConversationStats.PREVIEW_MAX_LEN
         sql = text(
             f"""
             SELECT
                 pme.conversation_id,
                 COUNT(DISTINCT pme.sequence) AS msg_count,
                 (
-                    SELECT SUBSTR(p2.converted_value, 1, {max_len + 3})
+                    SELECT SUBSTR(p2.converted_value, 1, {ConversationStats.PREVIEW_FETCH_MAX_LEN})
                     FROM "PromptMemoryEntries" p2
                     WHERE p2.conversation_id = pme.conversation_id
                     ORDER BY p2.sequence DESC, p2.id DESC
                     LIMIT 1
                 ) AS last_preview,
+                (
+                    SELECT p2b.converted_value_data_type
+                    FROM "PromptMemoryEntries" p2b
+                    WHERE p2b.conversation_id = pme.conversation_id
+                    ORDER BY p2b.sequence DESC, p2b.id DESC
+                    LIMIT 1
+                ) AS last_data_type,
                 (
                     SELECT p3.labels
                     FROM "PromptMemoryEntries" p3
@@ -774,11 +780,7 @@ def get_conversation_stats(self, *, conversation_ids: Sequence[str]) -> dict[str
 
         result: dict[str, ConversationStats] = {}
         for row in rows:
-            conv_id, msg_count, last_preview, raw_labels, raw_created_at = row
-
-            preview = None
-            if last_preview:
-                preview = last_preview[:max_len] + "..." if len(last_preview) > max_len else last_preview
+            conv_id, msg_count, last_preview, last_data_type, raw_labels, raw_created_at = row
 
             labels: dict[str, str] = {}
             if raw_labels and raw_labels not in ("null", "{}"):
@@ -794,7 +796,8 @@ def get_conversation_stats(self, *, conversation_ids: Sequence[str]) -> dict[str
 
             result[conv_id] = ConversationStats(
                 message_count=msg_count,
-                last_message_preview=preview,
+                last_message_preview=last_preview,
+                last_message_data_type=last_data_type,
                 labels=labels,
                 created_at=created_at,
             )

pyrit/models/__init__.py

@@ -62,7 +62,14 @@
     snake_case_to_class_name,
     validate_registry_name,
 )
-from pyrit.models.literals import ChatMessageRole, Modality, PromptDataType, PromptResponseError, SeedType
+from pyrit.models.literals import (
+    MEDIA_PATH_DATA_TYPES,
+    ChatMessageRole,
+    Modality,
+    PromptDataType,
+    PromptResponseError,
+    SeedType,
+)
 from pyrit.models.messages import (
     Message,
     MessagePiece,
@@ -141,6 +148,7 @@
     "IdentifierFilter",
     "IdentifierType",
     "ImagePathDataTypeSerializer",
+    "MEDIA_PATH_DATA_TYPES",
     "Message",
     "MessagePiece",
     "Modality",

pyrit/models/conversation_stats.py

@@ -6,6 +6,8 @@
 
 from pydantic import BaseModel, ConfigDict, Field
 
+from pyrit.models.literals import PromptDataType
+
 
 class ConversationStats(BaseModel):
     """
@@ -17,8 +19,17 @@ class ConversationStats(BaseModel):
     model_config = ConfigDict(frozen=True)
 
     PREVIEW_MAX_LEN: ClassVar[int] = 100
+    PREVIEW_FETCH_MAX_LEN: ClassVar[int] = 1024
+    """
+    Upper bound (in characters) for the raw ``last_message_preview`` value
+    fetched from storage. Larger than ``PREVIEW_MAX_LEN`` so that downstream
+    presentation code (see ``pyrit.backend.mappers._preview``) has enough
+    characters to extract a basename from a long media path or signed blob
+    URL before applying display-level truncation.
+    """
 
     message_count: int = 0
     last_message_preview: Optional[str] = None
+    last_message_data_type: Optional[PromptDataType] = None
     labels: dict[str, str] = Field(default_factory=dict)
     created_at: Optional[datetime] = None

pyrit/models/literals.py

@@ -18,6 +18,12 @@
     "function_call_output",
 ]
 
+# Subset of ``PromptDataType`` values whose stored ``value`` is a path or URL
+# pointing at media content (rather than the content itself). Useful for
+# treating these specially — e.g. avoiding raw filesystem-path leaks in API
+# previews, or signing blob storage URLs before exposing them to the frontend.
+MEDIA_PATH_DATA_TYPES: frozenset[PromptDataType] = frozenset({"image_path", "audio_path", "video_path", "binary_path"})
+
 """
 The type of the error in the prompt response
 blocked: blocked by an external filter e.g. Azure Filters