Merge branch 'main' into romanlutz-landing-raccoon-peek

Author: romanlutz

.pyrit_conf_example

@@ -46,6 +46,7 @@ memory_db_type: sqlite
 #         - scorer
 initializers:
   - name: simple
+  - name: scenario_technique
   - name: load_default_datasets
   - name: target
     args:

doc/code/converters/1_text_to_text_converters.ipynb

@@ -696,6 +696,11 @@
     "decomposition_converter = DecompositionConverter(converter_target=attack_llm)\n",
     "print(\"Decomposition:\", await decomposition_converter.convert_async(prompt=prompt))  # type: ignore\n",
     "\n",
+    "# With use_word_game=True, each noun phrase is also replaced by an innocuous codeword, with the\n",
+    "# mapping established in the same prompt\n",
+    "decomposition_word_game = DecompositionConverter(converter_target=attack_llm, use_word_game=True)\n",
+    "print(\"Decomposition (word-game):\", await decomposition_word_game.convert_async(prompt=prompt))  # type: ignore\n",
+    "\n",
     "# Denylist detection\n",
     "denylist_converter = DenylistConverter(converter_target=attack_llm)\n",
     "print(\"Denylist Check:\", await denylist_converter.convert_async(prompt=prompt))  # type: ignore\n",

doc/code/converters/1_text_to_text_converters.py

@@ -313,6 +313,11 @@
 decomposition_converter = DecompositionConverter(converter_target=attack_llm)
 print("Decomposition:", await decomposition_converter.convert_async(prompt=prompt))  # type: ignore
 
+# With use_word_game=True, each noun phrase is also replaced by an innocuous codeword, with the
+# mapping established in the same prompt
+decomposition_word_game = DecompositionConverter(converter_target=attack_llm, use_word_game=True)
+print("Decomposition (word-game):", await decomposition_word_game.convert_async(prompt=prompt))  # type: ignore
+
 # Denylist detection
 denylist_converter = DenylistConverter(converter_target=attack_llm)
 print("Denylist Check:", await denylist_converter.convert_async(prompt=prompt))  # type: ignore

pyproject.toml

@@ -209,6 +209,23 @@ include = ["pyrit/prompt_target/hugging_face/**"]
 [tool.ty.overrides.rules]
 invalid-argument-type = "ignore"
 
+# Unit tests intentionally exercise runtime validation failures, mutable test doubles,
+# and optional fields after fixtures have populated them. Keep library code strict while
+# avoiding hundreds of per-assertion ignores in tests.
+[[tool.ty.overrides]]
+include = ["tests/unit/**"]
+[tool.ty.overrides.rules]
+call-non-callable = "ignore"
+invalid-assignment = "ignore"
+invalid-await = "ignore"
+invalid-parameter-default = "ignore"
+invalid-type-arguments = "ignore"
+invalid-yield = "ignore"
+missing-argument = "ignore"
+not-subscriptable = "ignore"
+unknown-argument = "ignore"
+unsupported-operator = "ignore"
+
 [tool.uv]
 constraint-dependencies = [
     "aiohttp>=3.13.4",

pyrit/backend/models/__init__.py

@@ -7,6 +7,7 @@
 Pydantic models for API requests and responses.
 """
 
+from pyrit.backend.models._media import DEFAULT_MEDIA_EXTENSIONS
 from pyrit.backend.models.attacks import (
     AddMessageRequest,
     AddMessageResponse,
@@ -66,6 +67,8 @@
 )
 
 __all__ = [
+    # Media
+    "DEFAULT_MEDIA_EXTENSIONS",
     # Attacks
     "AddMessageRequest",
     "AddMessageResponse",

pyrit/backend/models/_media.py

@@ -29,8 +29,15 @@
     "binary_path": "file",
 }
 
-# Fallback extension per prefix when the value carries no usable suffix.
-_DEFAULT_EXTENSIONS = {"image": ".png", "audio": ".wav", "video": ".mp4", "file": ".bin"}
+# Default file extension per media data type, used when a value carries no usable
+# suffix and no MIME metadata is available. Centralized here so the backend response
+# models and the attack/converter services share a single source of truth.
+DEFAULT_MEDIA_EXTENSIONS: dict[str, str] = {
+    "image_path": ".png",
+    "audio_path": ".wav",
+    "video_path": ".mp4",
+    "binary_path": ".bin",
+}
 
 
 def infer_mime_type(*, value: str | None, data_type: PromptDataType) -> str | None:
@@ -79,6 +86,6 @@ def build_filename(*, data_type: str, sha256: str | None, value: str | None) ->
         ext = Path(source).suffix
 
     if not ext:
-        ext = _DEFAULT_EXTENSIONS.get(prefix, ".bin")
+        ext = DEFAULT_MEDIA_EXTENSIONS.get(data_type, ".bin")
 
     return f"{prefix}_{short_hash}{ext}"

pyrit/backend/services/attack_service.py

@@ -31,6 +31,7 @@
     request_piece_to_pyrit_message_piece,
     request_to_pyrit_message,
 )
+from pyrit.backend.models import DEFAULT_MEDIA_EXTENSIONS
 from pyrit.backend.models.attacks import (
     AddMessageRequest,
     AddMessageResponse,
@@ -935,22 +936,26 @@ async def _persist_base64_pieces_async(request: AddMessageRequest) -> None:
             except (OSError, ValueError):
                 pass
 
-            # Derive file extension from the MIME type sent by the frontend
-            ext = None
-            if piece.mime_type:
-                ext = mimetypes.guess_extension(piece.mime_type, strict=False)
-            if not ext:
-                ext = ".bin"
-
             # Strip data URI prefix if present (e.g. "data:image/png;base64,...")
             # The backend itself returns data URIs from pyrit_messages_to_dto_async,
             # so the client may echo them back.
             value = piece.original_value
+            data_uri_mime_type = None
             if value.startswith("data:"):
                 # Format: data:<mime>;base64,<payload>
-                _, _, payload = value.partition(",")
+                header, _, payload = value.partition(",")
+                data_uri_mime_type = header.split(":", 1)[1].split(";", 1)[0] if ":" in header else None
                 value = payload
 
+            # Derive file extension from MIME metadata, then fall back to data_type.
+            ext = None
+            if piece.mime_type:
+                ext = mimetypes.guess_extension(piece.mime_type, strict=False)
+            if not ext and data_uri_mime_type:
+                ext = mimetypes.guess_extension(data_uri_mime_type, strict=False)
+            if not ext:
+                ext = DEFAULT_MEDIA_EXTENSIONS.get(piece.data_type, ".bin")
+
             serializer = data_serializer_factory(
                 category="prompt-memory-entries",
                 data_type=cast("PromptDataType", piece.data_type),

pyrit/backend/services/converter_service.py

@@ -19,10 +19,11 @@
 import uuid
 from functools import lru_cache
 from pathlib import Path
-from typing import Any, ClassVar, Literal, Union, get_args, get_origin
+from typing import Any, Literal, Union, get_args, get_origin
 from urllib.parse import parse_qs, urlparse
 
 from pyrit.backend.mappers.converter_mappers import converter_object_to_instance
+from pyrit.backend.models import DEFAULT_MEDIA_EXTENSIONS
 from pyrit.backend.models.converters import (
     ConverterCatalogEntry,
     ConverterCatalogResponse,
@@ -79,13 +80,6 @@ class ConverterService:
     API metadata is derived from the converter objects.
     """
 
-    _DATA_TYPE_EXTENSION: ClassVar[dict[str, str]] = {
-        "image_path": ".png",
-        "audio_path": ".wav",
-        "video_path": ".mp4",
-        "binary_path": ".bin",
-    }
-
     def __init__(self) -> None:
         """Initialize the converter service."""
         self._registry = ConverterRegistry.get_registry_singleton()
@@ -254,7 +248,7 @@ async def preview_conversion_async(self, *, request: ConverterPreviewRequest) ->
             elif original_value.startswith("data:"):
                 _, _, value = original_value.partition(",")
 
-                ext = self._DATA_TYPE_EXTENSION.get(str(data_type), ".bin")
+                ext = DEFAULT_MEDIA_EXTENSIONS.get(str(data_type), ".bin")
 
                 serializer = data_serializer_factory(
                     category="prompt-memory-entries",
@@ -268,7 +262,7 @@ async def preview_conversion_async(self, *, request: ConverterPreviewRequest) ->
                 pass
             else:
                 # Treat as raw base64
-                ext = self._DATA_TYPE_EXTENSION.get(str(data_type), ".bin")
+                ext = DEFAULT_MEDIA_EXTENSIONS.get(str(data_type), ".bin")
 
                 serializer = data_serializer_factory(
                     category="prompt-memory-entries",

pyrit/cli/api_client.py

@@ -153,7 +153,7 @@ async def register_initializer_async(self, *, name: str, script_content: str) ->
             json={"name": name, "script_content": script_content},
         )
         if resp.status_code == 403:
-            detail = resp.json().get("detail", "Custom initializer operations are disabled on the server.")
+            detail = self._response_detail(resp) or "Custom initializer operations are disabled on the server."
             raise ServerNotAvailableError(detail)
         self._raise_for_status(resp)
         return resp.json()
@@ -308,6 +308,39 @@ async def _get_json_async(self, *, path: str, params: dict[str, Any] | None = No
         self._raise_for_status(resp)
         return resp.json()
 
+    @staticmethod
+    def _response_detail(resp: Any) -> str | None:
+        """
+        Extract a user-facing error detail from a response body.
+
+        Prefer FastAPI-style JSON ``detail`` values, then fall back to a plain
+        text response body. Non-string mock/proxy attributes are ignored so
+        callers can still use their default error messages.
+
+        Returns:
+            str | None: Extracted detail text, or ``None`` if the body has no
+                usable error detail.
+        """
+        try:
+            payload = resp.json()
+        except Exception:
+            payload = None
+        if isinstance(payload, dict):
+            detail_value = payload.get("detail")
+            if isinstance(detail_value, str) and detail_value.strip():
+                return detail_value
+            if detail_value is not None:
+                return str(detail_value)
+
+        text = getattr(resp, "text", "")
+        if isinstance(text, bytes):
+            text = text.decode(errors="replace")
+        if isinstance(text, str):
+            text = text.strip()
+            if text:
+                return text
+        return None
+
     @staticmethod
     def _raise_for_status(resp: Any) -> None:
         """
@@ -327,22 +360,7 @@ def _raise_for_status(resp: Any) -> None:
         try:
             resp.raise_for_status()
         except httpx.HTTPStatusError as exc:
-            detail: str | None = None
-            try:
-                payload = resp.json()
-            except Exception:
-                payload = None
-            if isinstance(payload, dict):
-                detail_value = payload.get("detail")
-                if isinstance(detail_value, str) and detail_value.strip():
-                    detail = detail_value
-                elif detail_value is not None:
-                    detail = str(detail_value)
-            if detail is None:
-                text = getattr(resp, "text", "") or ""
-                text = text.strip()
-                if text:
-                    detail = text
+            detail = PyRITApiClient._response_detail(resp)
             if detail is None:
                 raise
             message = f"{exc}: {detail}"