feat(guardrails): add IsMalicious MCP content gate

cookbook/guardrails/ismalicious/README.md

@@ -0,0 +1,41 @@
+# IsMalicious MCP guardrail
+
+Inspect normalized MCP argument strings before tool execution and text/structured-content strings before a tool result is returned. IsMalicious checks URL reputation for standalone HTTP(S) URL arguments, then scans the complete normalized text list for content injection and links
+
+## Configuration
+
+Obtain an API key and secret from [your account](https://ismalicious.com/app/account). In your secret manager, set `ISMALICIOUS_ENCODED_API_KEY` to the Base64 encoding of `apiKey:apiSecret`, with no newline. Do not log that value or put it directly into YAML
+
+Merge [`config.yaml`](config.yaml) into the proxy configuration containing your MCP servers. It sets both MCP modes and `default_on: true`. MCP subcalls do not necessarily inherit a parent chat request's guardrail selection, so relying only on a `guardrails` field on the parent request does not establish MCP enforcement
+
+```yaml
+guardrails:
+  - guardrail_name: ismalicious-mcp
+    litellm_params:
+      guardrail: ismalicious
+      mode: [pre_mcp_call, post_mcp_call]
+      default_on: true
+      api_key: os.environ/ISMALICIOUS_ENCODED_API_KEY
+```
+
+Requests only send credentials to `https://api.ismalicious.com`, with TLS verification, no redirects, a 15-second timeout and no retries. A custom API base is refused. Incoming request headers, provider credentials, user metadata and model prompts are not added to the inspection body by this provider
+
+## Decisions
+
+Top-level `warn` and `block` refuse. Timeout, quota429, redirects, invalid JSON/schema/verdict, incomplete link inspection and oversized serialized UTF-8 request bodies also refuse. A valid `allow` returns the original inputs without masking or sanitization. It means the service did not block under its current rules, not that content is proven benign. An unknown inner link reputation is a valid response and can still have an allow decision
+
+The content request serializes the complete normalized string list into `content` and uses `mode: fast`. The serialized HTTP body must fit within 1 MiB; it is never truncated. Results in `structuredContent`, including labels and numeric fields emitted by the native handler, are included when that handler exposes them. These calls use the separate scan quota, not the indicator lookup quota. URL reputation does not fetch the destination
+
+## Scope
+
+This provider uses LiteLLM's native MCP guardrail translation. It supports scannable text fields only. The native handler can skip a response that has no scannable text and can omit binary or multimodal blocks alongside text. This integration therefore does not enforce binary/multimodal/streaming content policy. Route only text tools through this policy and exclude unsupported tools at the gateway. Streaming LLM/tool effects, model-native search, hidden tool network calls and direct network access are outside its guarantees
+
+The normalized list is not the complete MCP envelope: transport metadata, content-block metadata, annotations, binary fields and fields not exposed by LiteLLM's string extraction are not inspected. Scanning this list must not be described as scanning every field in the original response
+
+The result can have existed in process memory or logging structures before the post-call inspection. Disable message/content logging and prompt storage as shown in the sample, do not install callbacks that expose raw tool output, and review your tracing configuration. Tool side effects cannot be undone. The failure message omits the raw text, URLs, credentials and upstream exception details. Normal MCP error responses may retain HTTP200 while indicating `isError`; clients must inspect the JSON-RPC/tool result rather than treating HTTP200 as an allow decision
+
+## Validation
+
+The mapped unit tests execute native pre/post MCP translation with an injected HTTP fixture. They check identity-preserving allowed inputs, original URL query/comma/fragment encoding, both refusal stages, structured-only results, warn, invalid verdicts, incomplete inspection, errors and UTF-8 limits without truncation. Synthetic replies validate integration policy, not the detector's accuracy
+
+An authenticated live proxy run against the actual IsMalicious API and a real LLM provider is still required before asking for maintainer review under this repository's contribution rules. The local MCP/unit tests must not be described as that proof. No pricing, plan counts or detection-accuracy claim is embedded in this example

cookbook/guardrails/ismalicious/config.yaml

@@ -0,0 +1,13 @@
+guardrails:
+  - guardrail_name: ismalicious-mcp
+    litellm_params:
+      guardrail: ismalicious
+      mode: [pre_mcp_call, post_mcp_call]
+      default_on: true
+      api_key: os.environ/ISMALICIOUS_ENCODED_API_KEY
+
+litellm_settings:
+  turn_off_message_logging: true
+
+general_settings:
+  store_prompts_in_spend_logs: false

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

@@ -0,0 +1,25 @@
+from typing import Final
+
+import litellm
+from litellm.types.guardrails import Guardrail, GuardrailEventHooks, LitellmParams, SupportedGuardrailIntegrations
+
+from .ismalicious import IsMaliciousGuardrail
+
+
+def initialize_guardrail(litellm_params: LitellmParams, guardrail: Guardrail) -> IsMaliciousGuardrail:
+    modes: Final = [litellm_params.mode] if isinstance(litellm_params.mode, str) else litellm_params.mode
+    if not isinstance(modes, list):
+        raise ValueError("IsMalicious requires explicit MCP modes")
+    callback: Final = IsMaliciousGuardrail(
+        api_key=litellm_params.api_key,
+        api_base=litellm_params.api_base,
+        guardrail_name=guardrail["guardrail_name"],
+        event_hook=[GuardrailEventHooks(mode) for mode in modes],
+        default_on=litellm_params.default_on is True,
+    )
+    litellm.logging_callback_manager.add_litellm_callback(callback)  # pyright: ignore[reportUnknownMemberType]  # manager exposes an untyped callable union
+    return callback
+
+
+guardrail_initializer_registry: Final = {SupportedGuardrailIntegrations.ISMALICIOUS.value: initialize_guardrail}
+guardrail_class_registry: Final = {SupportedGuardrailIntegrations.ISMALICIOUS.value: IsMaliciousGuardrail}

litellm/proxy/guardrails/guardrail_hooks/ismalicious/ismalicious.py

@@ -0,0 +1,216 @@
+import base64
+import binascii
+import json
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Final, Literal, NoReturn, Self
+from urllib.parse import urlsplit
+
+import httpx
+from pydantic import BaseModel, ConfigDict, Field, ValidationError, model_validator
+
+from litellm.exceptions import GuardrailRaisedException
+from litellm.integrations.custom_guardrail import CustomGuardrail
+from litellm.secret_managers.main import get_secret_str
+from litellm.types.guardrails import GuardrailEventHooks
+from litellm.types.utils import GenericGuardrailAPIInputs
+
+if TYPE_CHECKING:
+    from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
+    from litellm.types.proxy.guardrails.guardrail_hooks.base import GuardrailConfigModel
+
+_API_BASE: Final = "https://api.ismalicious.com"
+_MAX_BODY_BYTES: Final = 1024 * 1024
+_REFUSAL: Final = "IsMalicious could not allow the inspected MCP content"
+
+
+class _Response(BaseModel):
+    model_config = ConfigDict(extra="allow", strict=True)
+    verdict: Literal["allow", "warn", "block"]
+    latency_ms: int = Field(ge=0, le=2**63 - 1)
+
+
+class _Link(BaseModel):
+    model_config = ConfigDict(extra="allow", strict=True)
+    url: str
+    entity: str
+    verdict: Literal["clean", "suspicious", "malicious", "unknown"]
+    sources: int = Field(ge=0)
+
+
+class _Span(BaseModel):
+    model_config = ConfigDict(extra="allow", strict=True)
+    start: int = Field(ge=0)
+    end: int = Field(ge=0)
+    family: str
+
+    @model_validator(mode="after")
+    def ordered_interval(self) -> Self:
+        if self.end < self.start:
+            raise ValueError("Invalid injection span interval")
+        return self
+
+
+class _Injection(BaseModel):
+    model_config = ConfigDict(extra="allow", strict=True)
+    score: float = Field(ge=0, le=1)
+    families: list[str]
+    spans: list[_Span]
+
+
+class _Scan(_Response):
+    injection: _Injection
+    links: list[_Link]
+    links_truncated: bool
+    mode: Literal["fast", "thorough"]
+    source: _Link | None = None
+    sanitized_content: str | None = None
+
+
+class _Url(_Response):
+    url: str
+    entity: str
+    sources: int = Field(ge=0)
+
+
+@dataclass(frozen=True, slots=True)
+class _Failure:
+    blocked_content: bool = False
+
+
+def _verdict(response: httpx.Response, url: str | None) -> _Failure | None:
+    try:
+        response.raise_for_status()
+        parsed: Final = (_Scan if url is None else _Url).model_validate_json(response.content)
+    except (httpx.HTTPError, ValidationError, ValueError):
+        return _Failure()
+    if parsed.verdict != "allow":
+        return _Failure(blocked_content=True)
+    if isinstance(parsed, _Scan) and parsed.links_truncated:
+        return _Failure()
+    if isinstance(parsed, _Url) and parsed.url != url:
+        return _Failure()
+    return None
+
+
+def _encoded_body(texts: list[str]) -> bytes | _Failure:
+    try:
+        content: Final = json.dumps(texts, ensure_ascii=False, separators=(",", ":"))
+        body: Final = json.dumps({"content": content, "mode": "fast"}, ensure_ascii=False).encode("utf-8")
+    except (UnicodeError, ValueError):
+        return _Failure()
+    return body if len(body) <= _MAX_BODY_BYTES else _Failure()
+
+
+def _url_argument(text: str) -> str | None | _Failure:
+    if not text.startswith(("http://", "https://")):
+        return None
+    try:
+        parsed: Final = urlsplit(text)
+    except ValueError:
+        return _Failure()
+    if not parsed.hostname or parsed.username or parsed.password or any(ord(char) < 33 for char in text):
+        return _Failure()
+    return text
+
+
+class IsMaliciousGuardrail(CustomGuardrail):
+    def __init__(
+        self,
+        api_key: str | None = None,
+        api_base: str | None = None,
+        guardrail_name: str | None = None,
+        event_hook: GuardrailEventHooks | list[GuardrailEventHooks] | None = None,
+        default_on: bool = False,
+        transport: httpx.AsyncBaseTransport | None = None,
+    ) -> None:
+        resolved_key: Final = api_key or get_secret_str("ISMALICIOUS_ENCODED_API_KEY")
+        if not resolved_key:
+            raise ValueError("IsMalicious requires a Base64 API key and secret pair")
+        try:
+            decoded_key: Final = base64.b64decode(resolved_key, validate=True).decode("utf-8")
+        except (binascii.Error, UnicodeError):
+            raise ValueError("IsMalicious requires a Base64 API key and secret pair") from None
+        if ":" not in decoded_key or not all(decoded_key.split(":", 1)):
+            raise ValueError("IsMalicious requires a Base64 API key and secret pair")
+        if api_base is not None and api_base.rstrip("/") != _API_BASE:
+            raise ValueError("IsMalicious credentials may only be sent to its HTTPS API")
+        modes: Final = event_hook if isinstance(event_hook, list) else [event_hook]
+        if not modes or any(mode not in self.get_supported_event_hooks() for mode in modes):
+            raise ValueError("IsMalicious requires pre_mcp_call or post_mcp_call")
+        self._headers = {"X-API-KEY": resolved_key, "Content-Type": "application/json"}
+        self._transport = transport
+        super().__init__(  # pyright: ignore[reportUnknownMemberType]  # base guardrail accepts untyped kwargs
+            guardrail_name=guardrail_name,
+            supported_event_hooks=self.get_supported_event_hooks(),
+            event_hook=event_hook,
+            default_on=default_on,
+        )
+
+    @classmethod
+    def get_supported_event_hooks(cls) -> list[GuardrailEventHooks]:
+        return [GuardrailEventHooks.pre_mcp_call, GuardrailEventHooks.post_mcp_call]
+
+    @staticmethod
+    def get_config_model() -> type["GuardrailConfigModel[BaseModel]"] | None:
+        from litellm.types.proxy.guardrails.guardrail_hooks.ismalicious import IsMaliciousConfigModel
+
+        return IsMaliciousConfigModel
+
+    def _raise_failure(self, failure: _Failure) -> NoReturn:
+        raise GuardrailRaisedException(
+            guardrail_name=self.guardrail_name,
+            message=_REFUSAL,
+            blocked_content=failure.blocked_content,
+        )
+
+    async def _inspect(
+        self, client: httpx.AsyncClient, *, body: bytes | None = None, url: str | None = None
+    ) -> _Failure | None:
+        try:
+            response: Final = (
+                await client.get("/gate/url", params={"u": url})
+                if url is not None
+                else await client.post("/gate/scan", content=body)
+            )
+        except httpx.HTTPError:
+            return _Failure()
+        return _verdict(response, url)
+
+    async def _inspect_url_argument(self, client: httpx.AsyncClient, text: str) -> None:
+        url: Final = _url_argument(text)
+        if isinstance(url, _Failure):
+            self._raise_failure(url)
+        if isinstance(url, str):
+            failure: Final = await self._inspect(client, url=url)
+            if failure is not None:
+                self._raise_failure(failure)
+
+    async def apply_guardrail(
+        self,
+        inputs: GenericGuardrailAPIInputs,
+        request_data: dict[str, object],
+        input_type: Literal["request", "response"],
+        logging_obj: "LiteLLMLoggingObj | None" = None,
+    ) -> GenericGuardrailAPIInputs:
+        texts: Final = inputs.get("texts", [])
+        if inputs.get("images") or not texts:
+            self._raise_failure(_Failure())
+        body: Final = _encoded_body(texts)
+        if isinstance(body, _Failure):
+            self._raise_failure(body)
+        async with httpx.AsyncClient(
+            base_url=_API_BASE,
+            headers=self._headers,
+            transport=self._transport,
+            timeout=15,
+            follow_redirects=False,
+            verify=True,
+            trust_env=False,
+        ) as client:
+            if input_type == "request":
+                for text in texts:
+                    await self._inspect_url_argument(client, text)
+            scan_failure: Final = await self._inspect(client, body=body)
+            if scan_failure is not None:
+                self._raise_failure(scan_failure)
+        return inputs

litellm/types/guardrails.py

@@ -146,6 +146,7 @@ class SupportedGuardrailIntegrations(Enum):
     ALICE = "alice"
     AGENT_365 = "agent_365"
     CONDUCT = "conduct"
+    ISMALICIOUS = "ismalicious"
 
 
 class Role(Enum):

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

@@ -0,0 +1,11 @@
+from pydantic import BaseModel, Field
+
+from .base import GuardrailConfigModel
+
+
+class IsMaliciousConfigModel(GuardrailConfigModel[BaseModel]):
+    api_key: str | None = Field(default=None, description="Base64 encoding of the IsMalicious API key and secret pair")
+
+    @staticmethod
+    def ui_friendly_name() -> str:
+        return "IsMalicious"

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

@@ -0,0 +1,216 @@
+import base64
+import json
+
+import httpx
+import pytest
+from mcp.types import CallToolResult, TextContent
+
+from litellm.exceptions import GuardrailRaisedException
+from litellm.proxy._experimental.mcp_server.guardrail_translation.handler import MCPGuardrailTranslationHandler
+from litellm.proxy.guardrails.guardrail_hooks.ismalicious.ismalicious import IsMaliciousGuardrail
+from litellm.types.guardrails import GuardrailEventHooks
+
+URL = "https://example.com/path?q=a,b&x=1#part"
+KEY = base64.b64encode(b"test-key:test-secret").decode()
+
+
+def service_response(request, verdict="allow", truncated=False):
+    if request.url.path == "/gate/url":
+        body = {
+            "url": request.url.params["u"],
+            "entity": "example.com",
+            "verdict": verdict,
+            "sources": 0,
+            "latency_ms": 1,
+        }
+    else:
+        body = {
+            "verdict": verdict,
+            "injection": {"score": 0.0, "families": [], "spans": []},
+            "links": [{"url": URL, "entity": "example.com", "verdict": "unknown", "sources": 0}],
+            "links_truncated": truncated,
+            "mode": "fast",
+            "latency_ms": 1,
+        }
+    return httpx.Response(200, json=body)
+
+
+def guardrail(handler=service_response):
+    return IsMaliciousGuardrail(
+        api_key=KEY,
+        guardrail_name="ismalicious",
+        event_hook=[GuardrailEventHooks.pre_mcp_call, GuardrailEventHooks.post_mcp_call],
+        default_on=True,
+        transport=httpx.MockTransport(handler),
+    )
+
+
+@pytest.mark.asyncio
+async def test_native_pre_handler_preserves_original_url_and_arguments():
+    requests = []
+
+    def handler(request):
+        requests.append(request)
+        return service_response(request)
+
+    data = {"mcp_tool_name": "fetch", "mcp_arguments": {"url": URL, "other": "unchanged"}}
+    result = await MCPGuardrailTranslationHandler().process_input_messages(data, guardrail(handler))
+    assert result is data
+    assert requests[0].url.params["u"] == URL
+    assert json.loads(json.loads(requests[1].content)["content"]) == [URL, "unchanged"]
+    assert data["mcp_arguments"]["url"] == URL
+
+
+@pytest.mark.asyncio
+async def test_native_post_handler_preserves_text_and_structured_content_identity():
+    payload = CallToolResult(
+        content=[TextContent(type="text", text="Allowed å🚀 content")],
+        structuredContent={"context": "Additional untrusted context", "count": 3},
+    )
+    requests = []
+
+    def handler(request):
+        requests.append(request)
+        return service_response(request)
+
+    original_content = payload.content
+    original_structured = payload.structured_content
+    result = await MCPGuardrailTranslationHandler().process_output_response(payload, guardrail(handler))
+    assert result is payload
+    assert result.content is original_content
+    assert result.structured_content is original_structured
+    inspected = json.loads(json.loads(requests[0].content)["content"])
+    assert "Allowed å🚀 content" in inspected
+    assert "Additional untrusted context" in inspected
+    assert "context" in inspected and "count" in inspected and "3" in inspected
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("phase", ["url", "scan"])
+@pytest.mark.parametrize("verdict", ["warn", "block", "unknown"])
+async def test_native_pre_handler_does_not_return_a_blocked_request(phase, verdict):
+    def handler(request):
+        return service_response(request, verdict if request.url.path.endswith(phase) else "allow")
+
+    with pytest.raises(GuardrailRaisedException) as exc:
+        await MCPGuardrailTranslationHandler().process_input_messages(
+            {"mcp_tool_name": "fetch", "mcp_arguments": {"url": URL}}, guardrail(handler)
+        )
+    assert URL not in str(exc.value) and KEY not in str(exc.value)
+    assert exc.value.blocked_content is (verdict in {"warn", "block"})
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("verdict", ["warn", "block", "unknown"])
+async def test_native_post_handler_does_not_return_blocked_text(verdict):
+    payload = CallToolResult(content=[TextContent(type="text", text="Private untrusted result")])
+
+    def handler(request):
+        return service_response(request, verdict)
+
+    with pytest.raises(GuardrailRaisedException) as exc:
+        await MCPGuardrailTranslationHandler().process_output_response(payload, guardrail(handler))
+    assert "Private untrusted result" not in str(exc.value)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("failure", ["429", "302", "500", "bad-json", "missing", "timeout", "truncated"])
+async def test_native_post_handler_fails_closed_without_retry(failure):
+    calls = []
+
+    def handler(request):
+        calls.append(request)
+        if failure.isdigit():
+            return httpx.Response(int(failure), headers={"Location": "https://attacker.example"})
+        if failure == "bad-json":
+            return httpx.Response(200, content=b"not-json")
+        if failure == "missing":
+            return httpx.Response(200, json={"verdict": "allow"})
+        if failure == "timeout":
+            raise httpx.ReadTimeout("Sensitive detail", request=request)
+        return service_response(request, truncated=True)
+
+    with pytest.raises(GuardrailRaisedException) as exc:
+        await MCPGuardrailTranslationHandler().process_output_response(
+            CallToolResult(content=[TextContent(type="text", text="Untrusted result")]), guardrail(handler)
+        )
+    assert len(calls) == 1
+    assert not exc.value.blocked_content
+    assert "Sensitive detail" not in str(exc.value)
+
+
+@pytest.mark.asyncio
+async def test_serialized_utf8_request_body_limit_refuses_without_network():
+    def handler(request):
+        pytest.fail("An oversized request must not be sent")
+
+    with pytest.raises(GuardrailRaisedException):
+        await guardrail(handler).apply_guardrail(
+            inputs={"texts": ["🚀" * (1024 * 1024 // 4)]}, request_data={}, input_type="response"
+        )
+
+
+@pytest.mark.asyncio
+async def test_native_handler_scans_structured_only_untrusted_content():
+    def handler(request):
+        assert "Hidden instruction" in json.loads(request.content)["content"]
+        return service_response(request, "block")
+
+    with pytest.raises(GuardrailRaisedException):
+        await MCPGuardrailTranslationHandler().process_output_response(
+            CallToolResult(content=[], structuredContent={"instruction": "Hidden instruction"}), guardrail(handler)
+        )
+
+
+@pytest.mark.asyncio
+async def test_return_inputs_are_identity_without_rewriting():
+    inputs = {"texts": ["Allowed text"]}
+    assert await guardrail().apply_guardrail(inputs, {}, "response") is inputs
+
+
+@pytest.mark.parametrize("mode", [GuardrailEventHooks.pre_call, GuardrailEventHooks.during_mcp_call, None, []])
+def test_unsupported_modes_fail_at_startup(mode):
+    with pytest.raises(ValueError, match="requires pre_mcp_call"):
+        IsMaliciousGuardrail(api_key=KEY, event_hook=mode)
+
+
+@pytest.mark.parametrize("api_key", ["not-base64", "dXNlcg==", "OnNlY3JldA=="])
+def test_invalid_credentials_are_not_echoed(api_key):
+    with pytest.raises(ValueError, match="requires a Base64") as exc:
+        IsMaliciousGuardrail(api_key=api_key, event_hook=GuardrailEventHooks.pre_mcp_call)
+    assert api_key not in str(exc.value)
+
+
+def test_credentials_never_redirect_to_a_custom_endpoint():
+    with pytest.raises(ValueError, match="HTTPS API"):
+        IsMaliciousGuardrail(
+            api_key=KEY, api_base="https://attacker.example", event_hook=GuardrailEventHooks.pre_mcp_call
+        )
+
+
+def test_mcp_subcalls_without_guardrail_metadata_use_explicit_default_on():
+    gate = guardrail()
+    assert gate.should_run_guardrail(data={}, event_type=GuardrailEventHooks.pre_mcp_call)
+    assert gate.should_run_guardrail(data={}, event_type=GuardrailEventHooks.post_mcp_call)
+
+
+@pytest.mark.asyncio
+async def test_images_passed_to_the_provider_are_not_silently_allowed():
+    with pytest.raises(GuardrailRaisedException):
+        await guardrail().apply_guardrail({"texts": ["caption"], "images": ["image"]}, {}, "response")
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("invalid", ["backward-span", "latency-overflow"])
+async def test_malformed_response_semantics_fail_closed(invalid):
+    def handler(request):
+        body = service_response(request).json()
+        if invalid == "backward-span":
+            body["injection"]["spans"] = [{"start": 9, "end": 2, "family": "instruction"}]
+        else:
+            body["latency_ms"] = 2**63
+        return httpx.Response(200, json=body)
+
+    with pytest.raises(GuardrailRaisedException) as exc:
+        await guardrail(handler).apply_guardrail({"texts": ["Untrusted"]}, {}, "response")
+    assert not exc.value.blocked_content