fix(aicore): prevent content filter bypass in multi-module and streaming modes

src/sap_cloud_sdk/aicore/completion.py

@@ -5,15 +5,30 @@
 calls ``raise_for_status()`` on 4xx responses **before** our patched
 ``transform_response`` runs, so the 400 surfaces in user code as a
 ``litellm.APIConnectionError`` whose ``str(exc)`` contains the JSON body of
-the rejection. Output-filter rejections (HTTP 200 with
-``finish_reason == "content_filter"``) go through ``transform_response``
-and surface as :class:`ContentFilteredError`.
+the rejection. Output-filter rejections in buffered (non-streaming) mode
+(HTTP 200 with ``finish_reason == "content_filter"``) go through
+``transform_response`` and surface as :class:`ContentFilteredError`.
 
 That asymmetry would force callers to catch two exception types — these
 wrappers fix it by catching the wrapped exception inside the SDK and
 re-raising as :class:`ContentFilteredError` so callers can rely on a
 single exception type for "filter blocked you."
 
+Streaming + output filtering
+----------------------------
+When output filtering is active and ``stream=True``, the v2 API streams
+chunks normally and signals a filter rejection via an in-stream error event
+(no ``[DONE]``). LiteLLM's ``SAPStreamIterator`` surfaces this as a
+``ValueError`` raised from inside the iterator. These wrappers detect when
+output filtering is active and return a *filtering wrapper* around the
+iterator that catches that ``ValueError`` and re-raises it as
+:class:`ContentFilteredError`.
+
+Agent frameworks (``ChatLiteLLM``, LangGraph, etc.) that call
+``litellm.completion`` directly — bypassing this wrapper — will see the raw
+``ValueError`` from the iterator. They must use
+:func:`sap_cloud_sdk.aicore.completion` to get the typed :class:`ContentFilteredError`.
+
 Credential rotation handling
 ----------------------------
 When a credential (client_secret) is rotated while the pod is running,
@@ -40,16 +55,17 @@
         ...
 
 These are intentionally thin — every other keyword argument is forwarded
-verbatim to ``litellm.completion`` / ``litellm.acompletion``, including
-``stream=True`` (the streaming iterator is returned unchanged). No
-telemetry is recorded here; the wrappers fire on every LLM call and a
+verbatim to ``litellm.completion`` / ``litellm.acompletion``.
+
+No telemetry is recorded here; the wrappers fire on every LLM call and a
 counter at this level would be both noisy and uninformative about
 adoption of the SDK.
 """
 
 from __future__ import annotations
 
 import logging
+from collections.abc import AsyncIterator, Iterator
 from typing import Any
 
 import litellm
@@ -58,6 +74,8 @@
 from sap_cloud_sdk.core.telemetry.module import Module
 from sap_cloud_sdk.core.telemetry.operation import Operation
 from .filtering.filters import _parse_input_filter_error
+from .filtering import _patch as _filtering_patch
+from .filtering._patch import _parse_output_filter_stream_error
 
 logger = logging.getLogger(__name__)
 
@@ -84,45 +102,113 @@ def _maybe_translate_filter_error(exc: BaseException) -> BaseException:
     return blocked if blocked is not None else exc
 
 
+def _output_filtering_active() -> bool:
+    cfg = _filtering_patch._active_cfg
+    return cfg is not None and cfg.output_filtering is not None
+
+
+class _FilteringStreamWrapper:
+    """Sync iterator wrapper that converts in-stream output-filter errors."""
+
+    def __init__(self, stream: Iterator) -> None:
+        self._stream = stream
+
+    def __iter__(self) -> Iterator:
+        return self
+
+    def __next__(self) -> Any:
+        try:
+            return next(self._stream)
+        except StopIteration:
+            raise
+        except ValueError as exc:
+            translated = _parse_output_filter_stream_error(exc)
+            if translated is not None:
+                raise translated from exc
+            raise
+
+    def __getattr__(self, name: str) -> Any:
+        return getattr(self._stream, name)
+
+
+class _AsyncFilteringStreamWrapper:
+    """Async iterator wrapper that converts in-stream output-filter errors."""
+
+    def __init__(self, stream: AsyncIterator) -> None:
+        self._stream = stream
+
+    def __aiter__(self) -> AsyncIterator:
+        return self
+
+    async def __anext__(self) -> Any:
+        try:
+            return await self._stream.__anext__()
+        except StopAsyncIteration:
+            raise
+        except (ValueError, Exception) as exc:
+            translated = _parse_output_filter_stream_error(exc)
+            if translated is not None:
+                raise translated from exc
+            raise
+
+    def __getattr__(self, name: str) -> Any:
+        return getattr(self._stream, name)
+
+
 def completion(*args: Any, **kwargs: Any) -> Any:
     """Wrapper around :func:`litellm.completion` that normalises filter errors
     and handles credential rotation transparently.
 
     On ``AuthenticationError`` (e.g. rotated client_secret or mTLS cert),
     reloads credentials from the mounted secret volume and retries once.
 
+    When output filtering is active and ``stream=True``, wraps the returned
+    iterator so that in-stream output-filter rejections surface as
+    :class:`ContentFilteredError` rather than a raw ``ValueError``.
+
     Model strings (e.g. ``sap/<model>``) are passed verbatim to LiteLLM in all
     routing modes — proxy routing is handled by ``litellm.api_base`` configured
     in :func:`set_aicore_config`, not by rewriting the model name.
     """
+    is_streaming = kwargs.get("stream", False)
     try:
-        return litellm.completion(*args, **kwargs)
+        result = litellm.completion(*args, **kwargs)
     except litellm.AuthenticationError:
         _reload_reactive()
-        return litellm.completion(*args, **kwargs)
+        result = litellm.completion(*args, **kwargs)
     except Exception as exc:
         translated = _maybe_translate_filter_error(exc)
         if translated is exc:
             raise
         raise translated from exc
 
+    if is_streaming and _output_filtering_active():
+        return _FilteringStreamWrapper(result)
+    return result
+
 
 async def acompletion(*args: Any, **kwargs: Any) -> Any:
     """Async wrapper around :func:`litellm.acompletion`.
 
-    Same credential-rotation semantics as :func:`completion`.
+    Same credential-rotation and output-filter stream-wrapping semantics as
+    :func:`completion`.
     Model strings are passed verbatim to LiteLLM in all routing modes.
     """
+    is_streaming = kwargs.get("stream", False)
     try:
-        return await litellm.acompletion(*args, **kwargs)
+        result = await litellm.acompletion(*args, **kwargs)
     except litellm.AuthenticationError:
         _reload_reactive()
-        return await litellm.acompletion(*args, **kwargs)
+        result = await litellm.acompletion(*args, **kwargs)
     except Exception as exc:
         translated = _maybe_translate_filter_error(exc)
         if translated is exc:
             raise
         raise translated from exc
 
+    if is_streaming and _output_filtering_active():
+        return _AsyncFilteringStreamWrapper(result)
+    return result
+
 
 __all__ = ["completion", "acompletion"]

src/sap_cloud_sdk/aicore/filtering/_patch.py

@@ -4,14 +4,14 @@
 
 - Injects ``modules.filtering`` into every v2 completion request body via
   ``transform_request``.
-- Detects filter rejections in responses and raises ``ContentFilteredError``
-  via ``transform_response``.
+- Detects filter rejections in buffered responses and raises
+  ``ContentFilteredError`` via ``transform_response``.
 
 The patch is applied by :func:`_install` (called by ``set_filtering`` /
 ``disable_filtering`` in :mod:`.filters`) and is idempotent — calling it
 multiple times with the same config is safe.
 
-Two filter-rejection shapes (from the v2 API) are handled:
+Two filter-rejection shapes (from the v2 API) are handled in buffered mode:
 
 - Input rejection: HTTP 4xx with ``error.location`` startswith
   ``"Filtering Module - Input Filter"``.
@@ -26,10 +26,30 @@
 back into :class:`ContentFilteredError` before it reaches caller code; the
 :func:`_parse_input_filter_error` helper in :mod:`.filters` does the JSON
 parsing.
+
+Streaming + output filtering
+----------------------------
+The SAP AI Core Orchestration v2 API supports output filtering with streaming.
+The server filters each chunk and signals a rejection via an in-stream error
+event (no ``[DONE]``). LiteLLM's ``SAPStreamIterator`` surfaces this error
+event as a ``ValueError(json.dumps(event_obj))`` from inside the iterator.
+
+:func:`sap_cloud_sdk.aicore.completion` / :func:`acompletion` wrap the
+returned stream iterator when output filtering is active, catching that
+``ValueError`` and translating it to :class:`ContentFilteredError`. This gives
+callers a single, typed exception for filter rejections in both streaming and
+non-streaming modes.
+
+Note: agent frameworks (``ChatLiteLLM``, LangGraph, etc.) that call
+``litellm.completion`` directly — bypassing the SDK's completion wrapper —
+will see the raw ``ValueError`` from the iterator. They must either use
+:func:`sap_cloud_sdk.aicore.completion` to get the translated error, or
+handle the ``ValueError`` themselves.
 """
 
 from __future__ import annotations
 
+import json
 import logging
 from typing import Any
 
@@ -75,8 +95,8 @@ def transform_request(
 
         modules = body["config"]["modules"]
         if isinstance(modules, list):
-            # Fallback mode (list of configs) — inject into primary config only.
-            modules[0]["filtering"] = filtering_dict
+            for m in modules:
+                m["filtering"] = filtering_dict
         else:
             modules["filtering"] = filtering_dict
 
@@ -172,3 +192,46 @@ def _install(cfg: Any) -> None:  # cfg: ContentFiltering | None
     else:
         litellm.GenAIHubOrchestrationConfig = FilteringOrchestrationConfig
         logger.info("content filtering active (FilteringOrchestrationConfig)")
+
+
+def _parse_output_filter_stream_error(
+    exc: BaseException,
+) -> ContentFilteredError | None:
+    """Try to parse an in-stream output-filter rejection from a ValueError.
+
+    LiteLLM's ``SAPStreamIterator`` raises ``ValueError(json.dumps(event_obj))``
+    when the server sends an error event inside the stream.  The v2 error shape
+    for an output-filter rejection is::
+
+        {"error": {"request_id": "...", "code": 500,
+                   "message": "...",
+                   "location": "Filtering Module - Output Filter",
+                   "intermediate_results": {"output_filtering": {"data": {...}}}}}
+
+    Returns a :class:`ContentFilteredError` when the pattern matches, otherwise
+    ``None`` so the caller can re-raise the original exception unchanged.
+    """
+    if not isinstance(exc, (ValueError, Exception)):
+        return None
+    msg = str(exc)
+    brace = msg.find("{")
+    if brace == -1:
+        return None
+    try:
+        payload = json.loads(msg[brace:])
+    except (ValueError, TypeError):
+        return None
+    err = payload.get("error", {})
+    if not isinstance(err, dict):
+        return None
+    location = err.get("location") or ""
+    if not location.startswith("Filtering Module - Output Filter"):
+        return None
+    data = (
+        err.get("intermediate_results", {}).get("output_filtering", {}).get("data", {})
+    )
+    return ContentFilteredError(
+        direction="output",
+        details=data,
+        request_id=err.get("request_id"),
+    )

src/sap_cloud_sdk/aicore/filtering/exceptions.py

@@ -20,8 +20,11 @@ class ContentFilteredError(OrchestrationError):
     Attributes:
         direction: ``"input"`` if the user prompt was rejected (HTTP 4xx,
             ``error.location`` startswith ``"Filtering Module - Input Filter"``),
-            or ``"output"`` if the model response was rejected (HTTP 200,
-            ``finish_reason == "content_filter"``).
+            or ``"output"`` if the model response was rejected. In non-streaming
+            responses this is signalled by ``finish_reason == "content_filter"``
+            (HTTP 200). In streaming responses, the server sends an in-stream
+            error event which :func:`sap_cloud_sdk.aicore.completion` translates
+            into this exception from inside the iterator.
         details: Severity scalars from ``intermediate_results.{input,output}_filtering.data``.
             Safe to log; does NOT include raw prompt or completion content.
         request_id: ``error.request_id`` (input) or top-level ``request_id`` (output).

src/sap_cloud_sdk/aicore/user-guide.md

@@ -360,10 +360,12 @@ except ContentFilteredError as e:
     return "Your request was blocked by content safety policy."
 ```
 
-The wrapper forwards every argument verbatim to `litellm.completion`
-(including `stream=True`), and only intercepts the wrapped
-`APIConnectionError` shape that LiteLLM produces for input-filter
-rejections. All other exceptions surface unchanged.
+The wrapper forwards every argument verbatim to `litellm.completion`,
+including `stream=True`. Both input and output filtering work with streaming —
+filter rejections surface as `ContentFilteredError` regardless. See the
+[Streaming Responses](#streaming-responses) section for details.
+
+All other exceptions surface unchanged.
 
 `ContentFilteredError` exposes three attributes — `direction`, `details`,
 `request_id`. The `details` field contains severity scalars from the server,
@@ -444,21 +446,33 @@ print(response.choices[0].message.content)
 
 ### Streaming Responses
 
+`stream=True` is fully supported, including when output filtering is active.
+When the server rejects a streamed response via an in-stream filter event, the
+SDK raises `ContentFilteredError(direction="output")` from inside the iterator
+— the same exception type used for buffered (non-streaming) rejections.
+
+**Important:** this translation only applies when you use
+`sap_cloud_sdk.aicore.completion` / `acompletion`. If agent frameworks such as
+`ChatLiteLLM` or LangGraph call `litellm.completion` directly, mid-stream
+output-filter rejections surface as a raw `ValueError` from the iterator.
+
 ```python
-from sap_cloud_sdk.aicore import set_aicore_config
-from litellm import completion
+from sap_cloud_sdk.aicore import ContentFilteredError, completion, set_aicore_config
 
 set_aicore_config()
 
-response = completion(
-    model="sap/gpt-4",
-    messages=[{"role": "user", "content": "Tell me a story"}],
-    stream=True,
-)
+try:
+    response = completion(
+        model="sap/gpt-4",
+        messages=[{"role": "user", "content": "Tell me a story"}],
+        stream=True,
+    )
 
-for chunk in response:
-    if chunk.choices[0].delta.content:
-        print(chunk.choices[0].delta.content, end="")
+    for chunk in response:
+        if chunk.choices[0].delta.content:
+            print(chunk.choices[0].delta.content, end="")
+except ContentFilteredError as e:
+    print(f"Blocked ({e.direction}): {e.details}")
 ```
 
 ### Using Different Models