Merge 8200451e3c into 82ca10cc2d

litellm/proxy/guardrails/guardrail_hooks/panw_prisma_airs/__init__.py

@@ -17,14 +17,12 @@ def initialize_guardrail(litellm_params: "LitellmParams", guardrail: "Guardrail"
     if not guardrail_name:
         raise ValueError("PANW Prisma AIRS: guardrail_name is required")
 
-    _panw_callback: Final = PanwPrismaAirsHandler(
-        **{
-            **litellm_params.model_dump(exclude_unset=True),
-            "guardrail_name": guardrail_name,
-            "event_hook": litellm_params.mode,
-            "default_on": litellm_params.default_on or False,
-        }
-    )
+    litellm_params_dump: Final = litellm_params.model_dump(exclude_unset=True)
+    litellm_params_dump["guardrail_name"] = guardrail_name
+    litellm_params_dump["event_hook"] = litellm_params.mode
+    litellm_params_dump["default_on"] = litellm_params.default_on or False
+
+    _panw_callback: Final = PanwPrismaAirsHandler(**litellm_params_dump)
     litellm.logging_callback_manager.add_litellm_callback(_panw_callback)
 
     return _panw_callback

litellm/proxy/guardrails/guardrail_hooks/panw_prisma_airs/panw_prisma_airs.py

@@ -134,6 +134,20 @@ if TYPE_CHECKING:
     from litellm.types.proxy.guardrails.guardrail_hooks.base import GuardrailConfigModel
 
 
+def _parse_bool_setting(name: str, value: object) -> bool:
+    if isinstance(value, bool):
+        return value
+    if value is None:
+        return False
+    if isinstance(value, str):
+        from litellm.secret_managers.main import str_to_bool
+
+        parsed: Final = str_to_bool(value)
+        if parsed is not None:
+            return parsed
+    raise ValueError(f"PANW Prisma AIRS: {name} must be true or false, got {value!r}")
+
+
 class PanwPrismaAirsHandler(CustomGuardrail):
     """
     LiteLLM Built-in Guardrail for Palo Alto Networks Prisma AI Runtime Security (AIRS).
@@ -160,6 +174,7 @@ class PanwPrismaAirsHandler(CustomGuardrail):
     #: back would hand the caller exactly the text the operator declined to deliver.
     #: ``prompt_masked_data`` is deliberately NOT withheld: it is the caller's own input,
     #: and it is one of the fields the ticket asks for.
+    #: The ``include_masked_response_on_block`` config opt-in lifts the withholding.
     _CLIENT_HIDDEN_SCAN_FIELDS: Final = frozenset({"response_masked_data"})
 
     def __init__(
@@ -172,6 +187,7 @@ class PanwPrismaAirsHandler(CustomGuardrail):
         mask_on_block: bool = False,
         mask_request_content: bool = False,
         mask_response_content: bool = False,
+        include_masked_response_on_block: bool | str | None = False,
         app_name: str | None = None,
         fallback_on_error: Literal["block", "allow"] = "block",
         timeout: float = 10.0,
@@ -226,6 +242,11 @@ class PanwPrismaAirsHandler(CustomGuardrail):
 
         self.http_client = http_client
         self.fallback_on_error = fallback_on_error
+        # The dashboard persists some fields as strings and Pydantic extras pass
+        # them through unvalidated, so a bare "false" must not be treated as truthy.
+        self.include_masked_response_on_block: Final[bool] = _parse_bool_setting(
+            "include_masked_response_on_block", include_masked_response_on_block
+        )
         # Coerce defensively. The dashboard UI persists this field as a JSON
         # string, and Pydantic extras (the path that splats model_dump into
         # this handler) preserve whatever type the user supplied. A string
@@ -744,12 +765,15 @@ class PanwPrismaAirsHandler(CustomGuardrail):
             },
         )
 
+        hidden_fields: Final = (
+            frozenset[str]() if self.include_masked_response_on_block else self._CLIENT_HIDDEN_SCAN_FIELDS
+        )
         return {
             "error": {
                 **{
                     key: value
                     for key, value in scan_result.items()
-                    if not key.startswith("_") and key not in self._CLIENT_HIDDEN_SCAN_FIELDS
+                    if not key.startswith("_") and key not in hidden_fields
                 },
                 "message": error_msg,
                 "type": "guardrail_violation",
@@ -1544,9 +1568,10 @@ class PanwPrismaAirsHandler(CustomGuardrail):
             else:
                 # Tool calls now go out as ordinary prompt/response text, so a
                 # response-side scan reports the model's arguments under
-                # response_masked_data, which _CLIENT_HIDDEN_SCAN_FIELDS already
-                # withholds. prompt_masked_data is the caller's own input again and
-                # must keep reaching them -- it is one of the fields LIT-5638 asks for.
+                # response_masked_data, which _CLIENT_HIDDEN_SCAN_FIELDS withholds
+                # unless include_masked_response_on_block is on. prompt_masked_data is
+                # the caller's own input again and must keep reaching them -- it is one
+                # of the fields LIT-5638 asks for.
                 error_detail = self._build_error_detail(scan_result, is_response=is_response)
                 raise HTTPException(status_code=400, detail=error_detail)
 

litellm/proxy/guardrails/guardrail_initializers.py

@@ -282,6 +282,7 @@ def initialize_panw_prisma_airs(litellm_params, guardrail):
         mask_on_block=getattr(litellm_params, "mask_on_block", False),
         mask_request_content=getattr(litellm_params, "mask_request_content", False),
         mask_response_content=getattr(litellm_params, "mask_response_content", False),
+        include_masked_response_on_block=getattr(litellm_params, "include_masked_response_on_block", False),  # pyright: ignore[reportUnknownArgumentType]  # litellm_params is untyped in this function; pydantic extras require getattr
         app_name=getattr(litellm_params, "app_name", None),
         fallback_on_error=getattr(litellm_params, "fallback_on_error", "block"),
         # `timeout` is now declared on BaseLitellmParams (Optional[float] = None),

litellm/types/proxy/guardrails/guardrail_hooks/panw_prisma_airs.py

@@ -40,6 +40,11 @@ class PanwPrismaAirsGuardrailConfigModel(GuardrailConfigModel):
         description="Apply masking to responses that would be blocked. When True, masked content is returned to the user instead of blocking the response.",
     )
 
+    include_masked_response_on_block: bool = Field(
+        default=False,
+        description="When a response is blocked, include AIRS response_masked_data (masked text and pattern_detections) in the 400 error body. Has no effect when mask_response_content is on, because that returns the masked text with a 200 instead.",
+    )
+
     fallback_on_error: Literal["block", "allow"] = Field(
         default="block",
         description="Action when PANW API is unavailable (timeout, rate limit, network error): 'block' (default, maximum security) rejects requests; 'allow' (high availability) proceeds without scanning. Authentication and configuration errors always block.",

tests/integration/observability/test_panw_airs_block_masked_response.py

@@ -0,0 +1,351 @@
+"""LIT-9336: ``include_masked_response_on_block`` decides whether the 400 a PANW
+post_call block raises also carries AIRS ``response_masked_data`` (masked text
+plus ``pattern_detections``) back to the caller.
+
+The double returns allow on prompt scans and a customer-shaped block body on
+response scans, so every denial here is a post_call block.
+"""
+
+import json
+import uuid
+from collections.abc import Mapping
+from pathlib import Path
+from typing import Final
+
+import pytest
+import yaml
+from integration._support.client import Gateway, object_value
+from integration._support.process import owned_proxy
+from integration._support.wire import Reply, Request, wire_server
+from pydantic import JsonValue
+
+_MODEL_TEXT: Final = "sensitive generation"
+
+
+def _masked_data() -> dict[str, JsonValue]:
+    return {
+        "data": "Sample (fictional) values:\n\n- Email ID: `XXXXXXXXXXXXXXXXXXXXX`\n- SSN: `XXXXXXXXXXX`",
+        "pattern_detections": [
+            {"locations": [[72, 74], [76, 82]], "pattern": "License Plate Number - USA"},
+            {"locations": [[72, 74], [79, 82]], "pattern": "Credit Card CVV"},
+            {"locations": [[72, 82]], "pattern": "Driver License - US"},
+            {"locations": [[41, 61]], "pattern": "Internet - addr_email"},
+        ],
+    }
+
+
+def _block_scan(with_masked_data: bool = True) -> dict[str, JsonValue]:
+    scan_id: Final = uuid.uuid4().hex
+    session_id: Final = uuid.uuid4().hex
+    return {
+        "action": "block",
+        "category": "malicious",
+        "error": False,
+        "errors": [],
+        "profile_id": "a6ecc0c6-9aee-41bf-b967-22c690a8b989",
+        "profile_name": "DLP_PII_NexusOS",
+        "prompt_detected": {},
+        "report_id": "R" + scan_id,
+        "response_detected": {"dlp": True, "toxic_content": False, "ungrounded": False},
+        "scan_id": scan_id,
+        "session_id": session_id,
+        "source": "AI-Runtime-API",
+        "timeout": False,
+        "tool_detected": {},
+        "tr_id": session_id,
+        "transaction_id": "pan_" + uuid.uuid4().hex,
+        **({"response_masked_data": _masked_data()} if with_masked_data else {}),
+    }
+
+
+def _scanner(request: Request) -> Reply:
+    assert request.target == "/v1/scan/sync/request"
+    contents: Final = json.loads(request.body)["contents"][0]
+    if "response" in contents:
+        return Reply(body=json.dumps(_block_scan()).encode())
+    return Reply(
+        body=json.dumps({"action": "allow", "category": "benign", "response_detected": {}}).encode()
+    )
+
+
+def _scanner_without_masked_data(request: Request) -> Reply:
+    assert request.target == "/v1/scan/sync/request"
+    contents: Final = json.loads(request.body)["contents"][0]
+    if "response" in contents:
+        return Reply(body=json.dumps(_block_scan(with_masked_data=False)).encode())
+    return Reply(
+        body=json.dumps({"action": "allow", "category": "benign", "response_detected": {}}).encode()
+    )
+
+
+def _provider(request: Request) -> Reply:
+    if request.method == "GET" and request.target.endswith("/models"):
+        return Reply(body=json.dumps({"object": "list", "data": []}).encode())
+    if request.target == "/v1/chat/completions":
+        return Reply(
+            body=json.dumps(
+                {
+                    "id": "chatcmpl_synthetic",
+                    "object": "chat.completion",
+                    "created": 1700000000,
+                    "model": "gpt-4.1-mini",
+                    "choices": [
+                        {
+                            "index": 0,
+                            "finish_reason": "stop",
+                            "message": {"role": "assistant", "content": _MODEL_TEXT},
+                        }
+                    ],
+                    "usage": {"prompt_tokens": 5, "completion_tokens": 1, "total_tokens": 6},
+                }
+            ).encode()
+        )
+    if request.target == "/v1/responses":
+        return Reply(
+            body=json.dumps(
+                {
+                    "id": "resp_" + uuid.uuid4().hex,
+                    "object": "response",
+                    "created_at": 1700000000,
+                    "status": "completed",
+                    "model": "gpt-4.1-mini",
+                    "output": [
+                        {
+                            "type": "message",
+                            "id": "msg_synthetic",
+                            "status": "completed",
+                            "role": "assistant",
+                            "content": [{"type": "output_text", "text": _MODEL_TEXT, "annotations": []}],
+                        }
+                    ],
+                    "usage": {"input_tokens": 11, "output_tokens": 4, "total_tokens": 15},
+                }
+            ).encode()
+        )
+    if request.target == "/v1/messages":
+        return Reply(
+            body=json.dumps(
+                {
+                    "id": "msg_" + uuid.uuid4().hex,
+                    "type": "message",
+                    "role": "assistant",
+                    "model": "claude-sonnet-4-5-20250929",
+                    "content": [{"type": "text", "text": _MODEL_TEXT}],
+                    "stop_reason": "end_turn",
+                    "stop_sequence": None,
+                    "usage": {"input_tokens": 11, "output_tokens": 4},
+                }
+            ).encode()
+        )
+    raise AssertionError(f"unexpected provider target {request.target}")
+
+
+def _config(tmp_path: Path, identity: str, policy_url: str, **flags: bool) -> Path:
+    config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text())
+    config["guardrails"] = [
+        {
+            "guardrail_name": identity,
+            "litellm_params": {
+                "guardrail": "panw_prisma_airs",
+                "mode": "post_call",
+                "default_on": True,
+                "api_base": policy_url,
+                "api_key": "synthetic-panw-key",
+                "profile_name": "synthetic-profile",
+                **flags,
+            },
+        }
+    ]
+    path: Final = tmp_path / "panw_block.yaml"
+    path.write_text(yaml.safe_dump(config))
+    return path
+
+
+def _block_detail(node: JsonValue) -> Mapping[str, JsonValue] | None:
+    if isinstance(node, dict):
+        if node.get("code") == "panw_prisma_airs_response_blocked":
+            return node
+        for value in node.values():
+            found: Final = _block_detail(value)
+            if found is not None:
+                return found
+    if isinstance(node, list):
+        for value in node:
+            found: Final = _block_detail(value)
+            if found is not None:
+                return found
+    return None
+
+
+def _response_scans(requests: tuple[Request, ...]) -> list[dict[str, JsonValue]]:
+    return [
+        contents
+        for request in requests
+        for contents in json.loads(request.body)["contents"]
+        if "response" in contents
+    ]
+
+
+def _request_body(model: str, target: str) -> dict[str, JsonValue]:
+    if target == "/v1/responses":
+        return {"model": model, "input": "say hi"}
+    if target == "/v1/messages":
+        return {"model": model, "max_tokens": 16, "messages": [{"role": "user", "content": "say hi"}]}
+    return {"model": model, "messages": [{"role": "user", "content": "say hi"}]}
+
+
+@pytest.mark.parametrize("target", ["/v1/chat/completions", "/v1/responses", "/v1/messages"])
+def test_block_includes_response_masked_data_when_opted_in(gateway: Gateway, tmp_path: Path, target: str) -> None:
+    identity: Final = "guardrail" + uuid.uuid4().hex
+    with wire_server(_scanner) as policy, wire_server(_provider) as upstream:
+        config: Final = _config(tmp_path, identity, policy.url, include_masked_response_on_block=True)
+        with owned_proxy(gateway, tmp_path, {}, config=config) as candidate, candidate.scenario() as scenario:
+            model: Final = scenario.model(
+                model="openai/gpt-4.1-mini", api_base=upstream.url + "/v1", api_key="synthetic-key"
+            )
+            response: Final = candidate.request("POST", target, _request_body(model, target))
+
+            assert response.status_code == 400, response.text
+            detail: Final = _block_detail(object_value(response.json()))
+            assert detail is not None, f"blocked detail missing from {response.text}"
+            assert detail["action"] == "block"
+            assert detail["category"] == "malicious"
+            assert detail["response_masked_data"] == _masked_data()
+            assert _response_scans(policy.drain()) != [], "the AIRS double never saw a response scan"
+
+
+def test_block_withholds_response_masked_data_by_default(gateway: Gateway, tmp_path: Path) -> None:
+    identity: Final = "guardrail" + uuid.uuid4().hex
+    with wire_server(_scanner) as policy, wire_server(_provider) as upstream:
+        config: Final = _config(tmp_path, identity, policy.url)
+        with owned_proxy(gateway, tmp_path, {}, config=config) as candidate, candidate.scenario() as scenario:
+            model: Final = scenario.model(
+                model="openai/gpt-4.1-mini", api_base=upstream.url + "/v1", api_key="synthetic-key"
+            )
+            response: Final = candidate.request(
+                "POST", "/v1/chat/completions", _request_body(model, "/v1/chat/completions")
+            )
+
+            assert response.status_code == 400, response.text
+            detail: Final = _block_detail(object_value(response.json()))
+            assert detail is not None, f"blocked detail missing from {response.text}"
+            assert "response_masked_data" not in detail
+            assert detail["action"] == "block"
+            assert _response_scans(policy.drain()) != []
+
+
+def test_block_without_masked_data_invents_none(gateway: Gateway, tmp_path: Path) -> None:
+    identity: Final = "guardrail" + uuid.uuid4().hex
+    with wire_server(_scanner_without_masked_data) as policy, wire_server(_provider) as upstream:
+        config: Final = _config(tmp_path, identity, policy.url, include_masked_response_on_block=True)
+        with owned_proxy(gateway, tmp_path, {}, config=config) as candidate, candidate.scenario() as scenario:
+            model: Final = scenario.model(
+                model="openai/gpt-4.1-mini", api_base=upstream.url + "/v1", api_key="synthetic-key"
+            )
+            response: Final = candidate.request(
+                "POST", "/v1/chat/completions", _request_body(model, "/v1/chat/completions")
+            )
+
+            assert response.status_code == 400, response.text
+            detail: Final = _block_detail(object_value(response.json()))
+            assert detail is not None, f"blocked detail missing from {response.text}"
+            assert "response_masked_data" not in detail
+            assert _MODEL_TEXT not in response.text
+
+
+def test_mask_response_content_still_masks_when_flag_on(gateway: Gateway, tmp_path: Path) -> None:
+    identity: Final = "guardrail" + uuid.uuid4().hex
+    with wire_server(_scanner) as policy, wire_server(_provider) as upstream:
+        config: Final = _config(
+            tmp_path,
+            identity,
+            policy.url,
+            include_masked_response_on_block=True,
+            mask_response_content=True,
+        )
+        with owned_proxy(gateway, tmp_path, {}, config=config) as candidate, candidate.scenario() as scenario:
+            model: Final = scenario.model(
+                model="openai/gpt-4.1-mini", api_base=upstream.url + "/v1", api_key="synthetic-key"
+            )
+            response: Final = candidate.request(
+                "POST", "/v1/chat/completions", _request_body(model, "/v1/chat/completions")
+            )
+
+            assert response.status_code == 200, response.text
+            content: Final = response.json()["choices"][0]["message"]["content"]
+            assert content == _masked_data()["data"]
+
+
+def test_db_created_guardrail_string_false_withholds_response_masked_data_on_block(gateway: Gateway) -> None:
+    """The dashboard persists some litellm_params as strings; "false" must not turn the flag on."""
+    name: Final = f"panw-string-false-{uuid.uuid4().hex[:8]}"
+    with wire_server(_scanner) as policy, wire_server(_provider) as upstream:
+        gateway.post(
+            "/guardrails",
+            {
+                "guardrail": {
+                    "guardrail_name": name,
+                    "litellm_params": {
+                        "guardrail": "panw_prisma_airs",
+                        "mode": "post_call",
+                        "api_base": policy.url,
+                        "api_key": "synthetic-panw-key",
+                        "profile_name": "synthetic-profile",
+                        "include_masked_response_on_block": "false",
+                    },
+                }
+            },
+        )
+        with gateway.scenario() as scenario:
+            model: Final = scenario.model(
+                model="openai/gpt-4.1-mini", api_base=upstream.url + "/v1", api_key="synthetic-key"
+            )
+            response: Final = gateway.request(
+                "POST",
+                "/v1/chat/completions",
+                {**_request_body(model, "/v1/chat/completions"), "guardrails": [name]},
+            )
+
+            assert response.status_code == 400, response.text
+            detail: Final = _block_detail(object_value(response.json()))
+            assert detail is not None, f"blocked detail missing from {response.text}"
+            assert "response_masked_data" not in detail
+            assert _response_scans(policy.drain()) != [], "the AIRS double never saw a response scan"
+
+
+def test_db_created_guardrail_includes_response_masked_data_on_block(gateway: Gateway, tmp_path: Path) -> None:
+    name: Final = f"panw-block-details-{uuid.uuid4().hex[:8]}"
+    with wire_server(_scanner) as policy, wire_server(_provider) as upstream:
+        gateway.post(
+            "/guardrails",
+            {
+                "guardrail": {
+                    "guardrail_name": name,
+                    "litellm_params": {
+                        "guardrail": "panw_prisma_airs",
+                        "mode": "post_call",
+                        "api_base": policy.url,
+                        "api_key": "synthetic-panw-key",
+                        "profile_name": "synthetic-profile",
+                        "include_masked_response_on_block": True,
+                    },
+                }
+            },
+        )
+        with gateway.scenario() as scenario:
+            model: Final = scenario.model(
+                model="openai/gpt-4.1-mini", api_base=upstream.url + "/v1", api_key="synthetic-key"
+            )
+            response: Final = gateway.request(
+                "POST",
+                "/v1/chat/completions",
+                {**_request_body(model, "/v1/chat/completions"), "guardrails": [name]},
+            )
+
+            assert response.status_code == 400, response.text
+            detail: Final = _block_detail(object_value(response.json()))
+            assert detail is not None, f"blocked detail missing from {response.text}"
+            assert detail["action"] == "block"
+            assert detail["category"] == "malicious"
+            assert detail["response_masked_data"] == _masked_data()
+            assert _response_scans(policy.drain()) != [], "the AIRS double never saw a response scan"

tests/unit/proxy/guardrails/guardrail_hooks/test_panw_prisma_airs.py

@@ -6146,6 +6146,179 @@ class TestPanwAirsErrorDetailWithheldFields:
         assert detail["error"]["scan_id"] == "scan-2"
 
 
+class TestPanwAirsIncludeMaskedResponseOnBlock:
+    """LIT-9336: opt-in passthrough of AIRS ``response_masked_data`` in the 400 body.
+
+    With ``include_masked_response_on_block=True`` the withheld
+    ``_CLIENT_HIDDEN_SCAN_FIELDS`` lift, so the 400 detail carries the masked
+    text and ``pattern_detections`` exactly as AIRS returned them.
+    """
+
+    _CUSTOMER_BLOCK_RESPONSE = {
+        "action": "block",
+        "category": "malicious",
+        "error": False,
+        "errors": [],
+        "profile_id": "a6ecc0c6-9aee-41bf-b967-22c690a8b989",
+        "profile_name": "DLP_PII_NexusOS",
+        "prompt_detected": {},
+        "report_id": "Rec62eb82-78a5-43b6-b4fe-9e9dab03ff76",
+        "response_detected": {
+            "dlp": True,
+            "toxic_content": False,
+            "ungrounded": False,
+        },
+        "scan_id": "ec62eb82-78a5-43b6-b4fe-9e9dab03ff76",
+        "session_id": "2718c56e-a46d-47b1-afed-573c16c67d89",
+        "source": "AI-Runtime-API",
+        "timeout": False,
+        "tool_detected": {},
+        "tr_id": "2718c56e-a46d-47b1-afed-573c16c67d89",
+        "transaction_id": "pan_8e754890-5816-46dc-99ba-c881b7603a08",
+        "response_masked_data": {
+            "data": "Sample (fictional) values:\n\n- Email ID: `XXXXXXXXXXXXXXXXXXXXX`\n- SSN: `XXXXXXXXXXX`",
+            "pattern_detections": [
+                {"locations": [[72, 74], [76, 82]], "pattern": "License Plate Number - USA"},
+                {"locations": [[72, 74], [79, 82]], "pattern": "Credit Card CVV"},
+                {"locations": [[72, 82]], "pattern": "Driver License - US"},
+                {"locations": [[41, 61]], "pattern": "Internet - addr_email"},
+            ],
+        },
+    }
+
+    @pytest.mark.asyncio
+    async def test_block_includes_response_masked_data_when_opted_in(self):
+        handler = make_handler(include_masked_response_on_block=True)
+        inputs: GenericGuardrailAPIInputs = {"texts": ["Sensitive response data"]}
+        request_data = {"litellm_call_id": "test-call-id", "model": "gpt-4"}
+
+        with patch.object(handler, "_call_panw_api", new_callable=AsyncMock) as mock_api:
+            mock_api.return_value = copy.deepcopy(self._CUSTOMER_BLOCK_RESPONSE)
+
+            with pytest.raises(HTTPException) as exc_info:
+                await handler.apply_guardrail(
+                    inputs=inputs,
+                    request_data=request_data,
+                    input_type="response",
+                )
+
+        assert exc_info.value.status_code == 400
+        error = exc_info.value.detail["error"]
+
+        assert error["response_masked_data"] == self._CUSTOMER_BLOCK_RESPONSE["response_masked_data"]
+
+        for field in self._CUSTOMER_BLOCK_RESPONSE:
+            if field == "category":
+                continue
+            assert error[field] == self._CUSTOMER_BLOCK_RESPONSE[field], f"{field} missing or altered"
+
+        assert error["code"] == "panw_prisma_airs_response_blocked"
+        assert error["type"] == "guardrail_violation"
+        assert error["guardrail"] == "test_panw_airs"
+        assert error["category"] == "malicious"
+
+    @pytest.mark.asyncio
+    async def test_block_without_response_masked_data_stays_400(self):
+        """The opt-in passes through what AIRS sent; it does not invent the key."""
+        handler = make_handler(include_masked_response_on_block=True)
+        inputs: GenericGuardrailAPIInputs = {"texts": ["Sensitive response data"]}
+        request_data = {"litellm_call_id": "test-call-id", "model": "gpt-4"}
+
+        with patch.object(handler, "_call_panw_api", new_callable=AsyncMock) as mock_api:
+            mock_api.return_value = {
+                "action": "block",
+                "category": "malicious",
+                "scan_id": "scan-no-mask",
+            }
+
+            with pytest.raises(HTTPException) as exc_info:
+                await handler.apply_guardrail(
+                    inputs=inputs,
+                    request_data=request_data,
+                    input_type="response",
+                )
+
+        assert exc_info.value.status_code == 400
+        error = exc_info.value.detail["error"]
+        assert "response_masked_data" not in error
+        assert error["scan_id"] == "scan-no-mask"
+        assert error["code"] == "panw_prisma_airs_response_blocked"
+
+    @pytest.mark.asyncio
+    async def test_mask_response_content_still_masks_when_flag_on(self):
+        """mask_response_content wins over the flag: masked text goes out as a 200."""
+        handler = make_handler(
+            include_masked_response_on_block=True,
+            mask_response_content=True,
+        )
+        inputs: GenericGuardrailAPIInputs = {"texts": ["Sensitive response data"]}
+        request_data = {"litellm_call_id": "test-call-id", "model": "gpt-4"}
+
+        with patch.object(handler, "_call_panw_api", new_callable=AsyncMock) as mock_api:
+            mock_api.return_value = {
+                "action": "block",
+                "category": "dlp",
+                "response_masked_data": {"data": "XXXXXXXXX response data"},
+            }
+
+            result = await handler.apply_guardrail(
+                inputs=inputs,
+                request_data=request_data,
+                input_type="response",
+            )
+
+        assert result["texts"] == ["XXXXXXXXX response data"]
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("raw", ["false", "False", " false "])
+    async def test_string_false_withholds_response_masked_data(self, raw):
+        """A persisted "false" string must not be truthy on the withhold check."""
+        handler = make_handler(include_masked_response_on_block=raw)
+        inputs: GenericGuardrailAPIInputs = {"texts": ["Sensitive response data"]}
+        request_data = {"litellm_call_id": "test-call-id", "model": "gpt-4"}
+
+        with patch.object(handler, "_call_panw_api", new_callable=AsyncMock) as mock_api:
+            mock_api.return_value = copy.deepcopy(self._CUSTOMER_BLOCK_RESPONSE)
+
+            with pytest.raises(HTTPException) as exc_info:
+                await handler.apply_guardrail(
+                    inputs=inputs,
+                    request_data=request_data,
+                    input_type="response",
+                )
+
+        assert exc_info.value.status_code == 400
+        error = exc_info.value.detail["error"]
+        assert "response_masked_data" not in error
+        assert error["code"] == "panw_prisma_airs_response_blocked"
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("raw", ["true", "TRUE"])
+    async def test_string_true_includes_response_masked_data(self, raw):
+        handler = make_handler(include_masked_response_on_block=raw)
+        inputs: GenericGuardrailAPIInputs = {"texts": ["Sensitive response data"]}
+        request_data = {"litellm_call_id": "test-call-id", "model": "gpt-4"}
+
+        with patch.object(handler, "_call_panw_api", new_callable=AsyncMock) as mock_api:
+            mock_api.return_value = copy.deepcopy(self._CUSTOMER_BLOCK_RESPONSE)
+
+            with pytest.raises(HTTPException) as exc_info:
+                await handler.apply_guardrail(
+                    inputs=inputs,
+                    request_data=request_data,
+                    input_type="response",
+                )
+
+        assert exc_info.value.status_code == 400
+        error = exc_info.value.detail["error"]
+        assert error["response_masked_data"] == self._CUSTOMER_BLOCK_RESPONSE["response_masked_data"]
+
+    @pytest.mark.parametrize("raw", ["yes", 1, ""])
+    def test_non_boolean_value_fails_closed(self, raw):
+        with pytest.raises(ValueError, match="include_masked_response_on_block"):
+            make_handler(include_masked_response_on_block=raw)
+
+
 class TestPanwAirsToolCallBlockMaskedDataRouting:
     """A tool-call block must withhold model output and keep caller input.