REFACTOR: consolidate content-filter markers into shared constants

pyrit/exceptions/__init__.py

@@ -4,6 +4,7 @@
 """Exception classes, retry helpers, and execution context utilities."""
 
 from pyrit.exceptions.exception_classes import (
+    CONTENT_FILTER_MARKERS,
     BadRequestException,
     EmptyResponseException,
     ExperimentalWarning,
@@ -40,6 +41,7 @@
     "clear_execution_context",
     "clear_retry_collector",
     "ComponentRole",
+    "CONTENT_FILTER_MARKERS",
     "EmptyResponseException",
     "ExecutionContext",
     "ExecutionContextManager",

pyrit/exceptions/exception_classes.py

@@ -356,6 +356,33 @@ def pyrit_placeholder_retry(func: Callable[..., Any]) -> Callable[..., Any]:
     )(func)
 
 
+# Empirically-observed markers in OpenAI / Azure OpenAI / MAI error payloads that
+# indicate the response was blocked by a content filter or safety system.
+#
+# There is no canonical spec for these - providers expose the signal through
+# different field names (``error.code``, ``finish_reason``, ``incomplete_details.reason``,
+# free-form ``error.message``) and the exact wording evolves over time. Rather than
+# try to track every (provider, field) combination as an exact match, we scan the
+# entire payload as a substring search for resilience: adding support for a new
+# provider variant is then a one-line change to the set below.
+#
+# Each marker below is justified by a concrete provider response shape:
+#   - ``content_filter``           - OpenAI ``finish_reason``; Azure ``error.code``;
+#                                    Azure ``content_filter_results`` field name.
+#   - ``content_safety_violation`` - MAI image models ``error.code`` (added in PR #1890).
+#   - ``policy_violation``         - Substring of Azure's ``content_policy_violation``
+#                                    and OpenAI moderation's ``usage_policy_violation``.
+#   - ``moderation_blocked``       - OpenAI moderation ``error.code``.
+CONTENT_FILTER_MARKERS = frozenset(
+    {
+        "content_filter",
+        "content_safety_violation",
+        "policy_violation",
+        "moderation_blocked",
+    }
+)
+
+
 def handle_bad_request_exception(
     response_text: str,
     request: MessagePiece,
@@ -365,6 +392,11 @@ def handle_bad_request_exception(
     """
     Handle bad request responses and map them to standardized error messages.
 
+    The content-filter fallback substring-scans ``response_text`` against
+    ``CONTENT_FILTER_MARKERS`` so callers that do not pre-compute
+    ``is_content_filter`` (e.g. ``azure_ml_chat_target``) still benefit from
+    the full marker set.
+
     Args:
         response_text (str): Raw response text from the target.
         request (MessagePiece): Original request piece that caused the error.
@@ -378,11 +410,7 @@ def handle_bad_request_exception(
         RuntimeError: If the response does not match bad-request content-filter conditions.
 
     """
-    if (
-        "content_filter" in response_text
-        or "Invalid prompt: your prompt was flagged as potentially violating our usage policy." in response_text
-        or is_content_filter
-    ):
+    if is_content_filter or any(marker in response_text for marker in CONTENT_FILTER_MARKERS):
         # Handle bad request error when content filter system detects harmful content
         bad_request_exception = BadRequestException(status_code=error_code, message=response_text)
         resp_text = bad_request_exception.process_exception()

pyrit/prompt_target/openai/openai_error_handling.py

@@ -12,9 +12,30 @@
 import logging
 from typing import Optional, Union
 
+from pyrit.exceptions.exception_classes import CONTENT_FILTER_MARKERS
+
 logger = logging.getLogger(__name__)
 
 
+# OpenAI uses ``error.code == "invalid_prompt"`` for both model-level safety blocks
+# (e.g. CBRN topics) and unrelated failures (e.g. schema validation errors), so the
+# code alone is too generic to treat as a content-filter signal. Only treat
+# ``invalid_prompt`` as content filtering when the message text contains one of these
+# safety markers.
+#
+#   - "limited access"             - "we've limited access to this content for safety..."
+#   - "safety"                     - generic safety-system wording.
+#   - "usage policy"               - "your prompt was flagged as potentially violating
+#                                    our usage policy."
+SAFETY_MESSAGE_MARKERS = frozenset(
+    {
+        "limited access",
+        "safety",
+        "usage policy",
+    }
+)
+
+
 def _extract_request_id_from_exception(exc: Exception) -> Optional[str]:
     """
     Extract the x-request-id from an OpenAI SDK exception for logging/telemetry.
@@ -65,30 +86,26 @@ def _is_content_filter_error(data: Union[dict[str, object], str]) -> bool:
     """
     Check if error data indicates content filtering.
 
+    Performs a substring scan over the payload (JSON-dumped for dicts, ``str()`` for
+    strings) against ``CONTENT_FILTER_MARKERS``. The ``invalid_prompt`` code is
+    handled separately because it requires inspecting both the code and the message.
+
     Args:
         data: Either a dict (parsed JSON) or string (error text).
 
     Returns:
         True if content filtering is detected, False otherwise.
     """
     if isinstance(data, dict):
-        # Check for explicit content_filter or moderation_blocked codes
         error_obj = data.get("error")
-        code = error_obj.get("code") if isinstance(error_obj, dict) else None  # type: ignore[ty:invalid-argument-type]
-        if code in ["content_filter", "content_safety_violation", "moderation_blocked"]:
-            return True
-        # OpenAI uses "invalid_prompt" for model-level safety blocks (e.g. CBRN topics).
-        # Only treat it as a content filter when the message indicates a safety block,
-        # not for other invalid_prompt reasons (e.g. malformed schemas).
-        if code == "invalid_prompt":
-            message = error_obj.get("message", "") if isinstance(error_obj, dict) else ""  # type: ignore[ty:no-matching-overload]
-            if "limited access" in str(message).lower() or "safety" in str(message).lower():
+        if isinstance(error_obj, dict) and error_obj.get("code") == "invalid_prompt":  # type: ignore[ty:invalid-argument-type]
+            message = str(error_obj.get("message", "")).lower()  # type: ignore[ty:no-matching-overload]
+            if any(marker in message for marker in SAFETY_MESSAGE_MARKERS):
                 return True
-        # Heuristic: Azure sometimes uses other codes with policy-related content
-        return "content_filter" in json.dumps(data).lower()
-    # String-based heuristic search
-    lower = str(data).lower()
-    return "content_filter" in lower or "policy_violation" in lower or "moderation_blocked" in lower
+        haystack = json.dumps(data).lower()
+    else:
+        haystack = str(data).lower()
+    return any(marker in haystack for marker in CONTENT_FILTER_MARKERS)
 
 
 def _extract_error_payload(exc: Exception) -> tuple[Union[dict[str, object], str], bool]:
@@ -100,9 +117,10 @@ def _extract_error_payload(exc: Exception) -> tuple[Union[dict[str, object], str
     2. Fall back to e.body attribute
     3. Fall back to str(e)
 
-    It also attempts to detect whether the error is due to content filtering by:
-    - Checking for error.code == "content_filter"
-    - Searching for "content_filter" or "policy_violation" keywords in the payload
+    It also attempts to detect whether the error is due to content filtering by
+    delegating to ``_is_content_filter_error``, which scans the payload for the
+    markers in ``CONTENT_FILTER_MARKERS`` (or, for ``invalid_prompt`` errors,
+    inspects the message for ``SAFETY_MESSAGE_MARKERS``).
 
     Args:
         exc: An exception from the OpenAI SDK (typically BadRequestError).

tests/unit/exceptions/test_exceptions.py

@@ -6,17 +6,21 @@
 import os
 from contextlib import suppress
 
+import pytest
 from tenacity import RetryError
 
 from pyrit.exceptions import (
+    CONTENT_FILTER_MARKERS,
     BadRequestException,
     EmptyResponseException,
     InvalidJsonException,
     MissingPromptPlaceholderException,
     PyritException,
     RateLimitException,
+    handle_bad_request_exception,
     pyrit_custom_result_retry,
 )
+from pyrit.models import MessagePiece
 
 
 def test_pyrit_exception_initialization():
@@ -104,6 +108,66 @@ def test_remove_markdown_json_exception(caplog):
     assert "InvalidJsonException encountered: Status Code: 500, Message: Invalid JSON Response" in caplog.text
 
 
+def _make_request_piece() -> MessagePiece:
+    return MessagePiece(role="user", conversation_id="test-convo", original_value="hello")
+
+
+def test_content_filter_markers_exported_from_pyrit_exceptions():
+    """The marker set must be importable from ``pyrit.exceptions`` as the single source of truth."""
+    assert "content_filter" in CONTENT_FILTER_MARKERS
+    assert "moderation_blocked" in CONTENT_FILTER_MARKERS
+    assert "policy_violation" in CONTENT_FILTER_MARKERS
+    assert "content_safety_violation" in CONTENT_FILTER_MARKERS
+
+
+@pytest.mark.parametrize(
+    "marker_response_text",
+    [
+        "content_filter",
+        '{"error": {"code": "moderation_blocked"}}',
+        '{"error": {"code": "content_policy_violation"}}',
+        '{"error": {"code": "content_safety_violation"}}',
+    ],
+)
+def test_handle_bad_request_exception_returns_blocked_for_any_marker(marker_response_text):
+    """The substring fallback must trigger for every marker in ``CONTENT_FILTER_MARKERS``."""
+    try:
+        raise RuntimeError("simulated upstream error")
+    except RuntimeError:
+        response = handle_bad_request_exception(
+            response_text=marker_response_text,
+            request=_make_request_piece(),
+        )
+
+    assert response.message_pieces[0].response_error == "blocked"
+
+
+def test_handle_bad_request_exception_reraises_when_no_marker_and_not_content_filter():
+    """If neither ``is_content_filter`` nor any marker matches, the original exception must propagate."""
+    with pytest.raises(RuntimeError, match="simulated upstream error"):
+        try:
+            raise RuntimeError("simulated upstream error")
+        except RuntimeError:
+            handle_bad_request_exception(
+                response_text='{"error": {"code": "schema_validation_failed"}}',
+                request=_make_request_piece(),
+            )
+
+
+def test_handle_bad_request_exception_returns_blocked_when_is_content_filter_true():
+    """An explicit ``is_content_filter`` signal must trigger the blocked path regardless of response_text."""
+    try:
+        raise RuntimeError("simulated upstream error")
+    except RuntimeError:
+        response = handle_bad_request_exception(
+            response_text="some unrelated text without any marker",
+            request=_make_request_piece(),
+            is_content_filter=True,
+        )
+
+    assert response.message_pieces[0].response_error == "blocked"
+
+
 class TestRetryDecoratorsRespectRuntimeEnvVars:
     """
     Tests that retry decorators read environment variables at runtime, not at decoration time.

tests/unit/prompt_target/target/test_openai_error_handling.py

@@ -1,27 +1,68 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT license.
 
+import pytest
+
+from pyrit.exceptions import CONTENT_FILTER_MARKERS
 from pyrit.prompt_target.openai.openai_error_handling import (
+    SAFETY_MESSAGE_MARKERS,
     _is_content_filter_error,
 )
 
 # Tests for _is_content_filter_error helper
 
 
-def test_is_content_filter_error_with_dict():
-    """Test detection with dict input"""
-    data = {"error": {"code": "content_filter"}}
+def test_content_filter_markers_contents():
+    """Sanity-check the empirical marker set so accidental removals are caught."""
+    assert {
+        "content_filter",
+        "content_safety_violation",
+        "policy_violation",
+        "moderation_blocked",
+    } <= CONTENT_FILTER_MARKERS
+
+
+def test_safety_message_markers_contents():
+    """Sanity-check the message-level safety markers used for invalid_prompt."""
+    assert {"limited access", "safety", "usage policy"} <= SAFETY_MESSAGE_MARKERS
+
+
+@pytest.mark.parametrize(
+    "code",
+    [
+        "content_filter",
+        "content_safety_violation",
+        "moderation_blocked",
+    ],
+)
+def test_is_content_filter_error_explicit_code(code):
+    """Each marker that appears as an exact error.code should be detected."""
+    assert _is_content_filter_error({"error": {"code": code}}) is True
+
+
+def test_is_content_filter_error_content_policy_violation_via_substring():
+    """Azure's content_policy_violation code is detected via the policy_violation marker."""
+    data = {"error": {"code": "content_policy_violation", "message": "Content blocked"}}
     assert _is_content_filter_error(data) is True
 
 
+def test_is_content_filter_error_with_dict():
+    """Dict input with a content_filter code is detected."""
+    assert _is_content_filter_error({"error": {"code": "content_filter"}}) is True
+
+
 def test_is_content_filter_error_with_string():
-    """Test detection with string input containing content_filter"""
-    error_str = '{"error": {"code": "content_filter"}}'
-    assert _is_content_filter_error(error_str) is True
+    """String input containing a marker is detected."""
+    assert _is_content_filter_error('{"error": {"code": "content_filter"}}') is True
+
+
+def test_is_content_filter_error_string_moderation_blocked():
+    """String input containing moderation_blocked is detected."""
+    assert _is_content_filter_error("error: moderation_blocked for prompt") is True
 
 
 def test_is_content_filter_error_invalid_prompt_safety_block():
-    """Test detection with invalid_prompt code and safety-related message (CBRN block)"""
+    """invalid_prompt + 'safety' / 'limited access' message is detected (CBRN block)."""
     data = {
         "error": {
             "code": "invalid_prompt",
@@ -31,13 +72,40 @@ def test_is_content_filter_error_invalid_prompt_safety_block():
     assert _is_content_filter_error(data) is True
 
 
+def test_is_content_filter_error_invalid_prompt_usage_policy_message():
+    """invalid_prompt + 'usage policy' message is detected (previously a hardcoded literal)."""
+    data = {
+        "error": {
+            "code": "invalid_prompt",
+            "message": "Invalid prompt: your prompt was flagged as potentially violating our usage policy.",
+        }
+    }
+    assert _is_content_filter_error(data) is True
+
+
 def test_is_content_filter_error_invalid_prompt_non_safety():
-    """Test that invalid_prompt without a safety message is NOT treated as a content filter"""
+    """invalid_prompt without a safety-marker message is NOT treated as content filter."""
     data = {"error": {"code": "invalid_prompt", "message": "Invalid prompt: schema validation failed."}}
     assert _is_content_filter_error(data) is False
 
 
+def test_is_content_filter_error_invalid_prompt_non_safety_with_content_filter_marker():
+    """invalid_prompt with no safety message but a CONTENT_FILTER_MARKERS substring elsewhere is detected."""
+    data = {
+        "error": {
+            "code": "invalid_prompt",
+            "message": "Invalid prompt.",
+            "inner_error": {"code": "content_filter"},
+        }
+    }
+    assert _is_content_filter_error(data) is True
+
+
 def test_is_content_filter_error_no_filter():
-    """Test detection returns False when no content_filter"""
-    error_dict = {"error": {"code": "rate_limit", "message": "Too many requests"}}
-    assert _is_content_filter_error(error_dict) is False
+    """Unrelated errors return False."""
+    assert _is_content_filter_error({"error": {"code": "rate_limit", "message": "Too many requests"}}) is False
+
+
+def test_is_content_filter_error_string_no_filter():
+    """String input without any marker returns False."""
+    assert _is_content_filter_error("connection timed out") is False