scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-06 02:48:13 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 103

feat(guardrails): add Singulr guardrail integration for LiteLLM gateway (#31302)

Browse source 
* singulr guardrail support for litellm gateway

* Update litellm/proxy/guardrails/guardrail_hooks/singulr/singulr.py

Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>

* fix comments

* improvement

* fix: resolve review comments and implement requested improvements

* fix:Guardrail bypass through uninspected messages

* fix:tool text scanning

* fix: Legacy function definitions bypass scanning by adding indirect message scaning

* chore: remove unintended basedpyright budget file

* fix:Response schema bypasses guardrail scanning (response_format.json_schema)

* chore: restore basedpyright-code-budget.json and update lint baselines

Restores the file deleted in c698b88686 to match upstream litellm_internal_staging.
Regenerates basedpyright and ruff-strict budget baselines via make lint-budget-update.

* fix: scan system messages as indirect prompt injection in Singulr guardrail

* chore: restore lint budget files to upstream baseline

* fix: resolve ruff UP006 and I001 violations in singulr guardrail

* Update litellm/proxy/guardrails/guardrail_hooks/singulr/singulr.py

Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>

* resolve review comments on Singulr guardrail

* fix: scan tool call results as indirect prompt injection in Singulr guardrail

* Apply suggestion from @greptile-apps[bot]

Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>

* minor

* formating fix

* refactor: shift extraction logic to singulr side

* refactor:keep precall hook only

* fix:formatting

* fix:linting

* improve config description

* Trigger CI

* fix

* fix:field description

* fix:errors due to change in field names

* style: apply ruff line-wrap formatting to singulr guardrail

* fix:exception

* fix:formatting

* fix playground

* improved

* Update litellm/proxy/guardrails/guardrail_hooks/singulr/singulr.py

Co-authored-by: veria-ai[bot] <224490171+veria-ai[bot]@users.noreply.github.com>

* Update litellm/proxy/guardrails/guardrail_hooks/singulr/singulr.py

Co-authored-by: veria-ai[bot] <224490171+veria-ai[bot]@users.noreply.github.com>

* fix

* fix ci issues

* remove uv.lock from pr

* fix

* fix:resolved comments

* chore: trigger CI

* remove uv.lock

* fix

* fix linting

* fix linting

* fix linting

* remove doc strings

* remove test fixes

* chore: retrigger CI

* change in singulr api contract

* remove some ut

* send litellm call_id to singulr

---------

Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>
Co-authored-by: aniket-kardile <aniket.kardile@singulr.ai>
Co-authored-by: veria-ai[bot] <224490171+veria-ai[bot]@users.noreply.github.com>
This commit is contained in:
madan-singulr 2026-07-17 21:45:20 +05:30 • committed by GitHub
parent 69a476f791
commit 68cd817953
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 884 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

50
litellm/proxy/guardrails/guardrail_hooks/singulr/__init__.py Normal file
View file

@ -0,0 +1,50 @@
"""
Author: Madan Singhal
Date: 23/06/26
"""
from typing import TYPE_CHECKING
from litellm.types.guardrails import SupportedGuardrailIntegrations
from .singulr import SingulrGuardrail
if TYPE_CHECKING:
from litellm.types.guardrails import Guardrail, LitellmParams
def initialize_guardrail(
litellm_params: "LitellmParams",
guardrail: "Guardrail",
):
import litellm
_cb = SingulrGuardrail(
singulr_api_base=getattr(litellm_params, "singulr_api_base", None) or litellm_params.api_base,
singulr_api_key=getattr(litellm_params, "singulr_api_key", None) or litellm_params.api_key,
singulr_application_id=getattr(litellm_params, "singulr_application_id", None),
singulr_guardrail_id=getattr(litellm_params, "singulr_guardrail_id", None),
block_on_error=getattr(litellm_params, "block_on_error", None),
timeout=litellm_params.timeout,
guardrail_name=guardrail.get(
"guardrail_name",
"",
),
event_hook=litellm_params.mode,
default_on=litellm_params.default_on,
)
litellm.logging_callback_manager.add_litellm_callback(
_cb,
)
return _cb
guardrail_initializer_registry = {
SupportedGuardrailIntegrations.SINGULR.value: initialize_guardrail,
}
guardrail_class_registry = {
SupportedGuardrailIntegrations.SINGULR.value: SingulrGuardrail,
}

216
litellm/proxy/guardrails/guardrail_hooks/singulr/singulr.py Normal file
View file

@ -0,0 +1,216 @@
import os
from typing import Any
from urllib.parse import urlparse
import httpx
import pydantic
from litellm._logging import verbose_proxy_logger
from litellm.exceptions import GuardrailRaisedException
from litellm.integrations.custom_guardrail import (
CustomGuardrail,
log_guardrail_information,
)
from litellm.litellm_core_utils.litellm_logging import (
Logging as LiteLLMLoggingObj,
)
from litellm.llms.custom_httpx.http_handler import (
get_async_httpx_client,
httpxSpecialProvider,
)
from litellm.types.guardrails import GuardrailEventHooks
from litellm.types.proxy.guardrails.guardrail_hooks.base import (
GuardrailConfigModel,
)
from litellm.types.proxy.guardrails.guardrail_hooks.singulr import (
SingulrGuardrailPayload,
SingulrGuardrailRequest,
SingulrGuardrailResponse,
)
from litellm.types.utils import GenericGuardrailAPIInputs
_DEFAULT_API_BASE = "http://localhost:8003"
_GUARD_ENDPOINT = "/api/v1/ai-gateway/litellm"
_DEFAULT_TIMEOUT = 30.0
class SingulrGuardrail(CustomGuardrail):
def __init__(
self,
singulr_api_key: str | None = None,
singulr_api_base: str | None = None,
singulr_application_id: str | None = None,
singulr_guardrail_id: str | None = None,
block_on_error: bool | None = None,
timeout: float | None = None,
**kwargs: Any,
) -> None:
self.singulr_api_key = singulr_api_key or os.environ.get("SINGULR_API_KEY")
self.singulr_api_base = (singulr_api_base or os.environ.get("SINGULR_API_BASE") or _DEFAULT_API_BASE).rstrip(
"/"
)
parsed = urlparse(self.singulr_api_base)
if parsed.scheme == "http" and parsed.hostname not in (
"localhost",
"127.0.0.1",
):
raise ValueError(
f"Singulr: api_base {self.singulr_api_base} uses plain HTTP for a "
"non-local endpoint. Guardrail payloads contain the API token, full "
"conversation content, and the guardrail decision, so this endpoint "
"must use HTTPS."
)
self.singulr_application_id = singulr_application_id or os.environ.get("SINGULR_ENFORCEMENT_ENTITY_ID")
self.singulr_guardrail_id = singulr_guardrail_id or os.environ.get("SINGULR_GUARDRAIL_ID")
if block_on_error is None:
env = os.environ.get("SINGULR_BLOCK_ON_ERROR", "true")
self.block_on_error = env.lower() in ("true", "1", "yes")
else:
self.block_on_error = block_on_error
self.timeout = _DEFAULT_TIMEOUT if timeout is None else timeout
self.async_handler = get_async_httpx_client(
llm_provider=httpxSpecialProvider.GuardrailCallback,
)
if "supported_event_hooks" not in kwargs:
kwargs["supported_event_hooks"] = [
GuardrailEventHooks.pre_call,
GuardrailEventHooks.post_call,
]
super().__init__(**kwargs)
@staticmethod
def get_config_model() -> type["GuardrailConfigModel"] | None:
from litellm.types.proxy.guardrails.guardrail_hooks.singulr import (
SingulrGuardrailConfigModel,
)
return SingulrGuardrailConfigModel
def _build_payload(
self,
request_data: dict[str, Any],
inputs: GenericGuardrailAPIInputs,
input_type: str,
) -> dict[str, Any]:
if not request_data:
texts = inputs.get("texts", [])
payload = SingulrGuardrailPayload(
input_type=input_type,
is_playground_request=True,
playground_text=texts[0] if texts else None,
)
else:
response = request_data.get("response")
singulr_req_object = SingulrGuardrailRequest(
model=request_data.get("model"),
messages=request_data.get("messages"),
tools=request_data.get("tools"),
model_response=response.model_dump(mode="json") if input_type == "response" and response else None,
litellm_metadata=request_data.get("litellm_metadata"),
)
payload = SingulrGuardrailPayload(
litellm_call_id=request_data.get("litellm_call_id"),
request_data=singulr_req_object,
input_type=input_type,
)
return payload.model_dump(mode="json")
def _build_headers(self) -> dict[str, str]:
return dict(
(header, value)
for header, value in (
("Content-Type", "application/json"),
("X-Singulr-Gateway-Token", self.singulr_api_key),
(
"X-Singulr-Enforcement-Entity-Id",
self.singulr_application_id or "",
),
("X-Singulr-Guardrail-Id", self.singulr_guardrail_id or ""),
)
if value
)
async def _call_api(self, payload: dict[str, Any]) -> SingulrGuardrailResponse | None:
endpoint = f"{self.singulr_api_base}{_GUARD_ENDPOINT}"
verbose_proxy_logger.debug("Singulr: %s", endpoint)
try:
response = await self.async_handler.post(
url=endpoint,
headers=self._build_headers(),
json=payload,
timeout=self.timeout,
)
response.raise_for_status()
result = SingulrGuardrailResponse.model_validate(response.json())
verbose_proxy_logger.debug("Singulr: result=%s", result)
return result
except httpx.HTTPStatusError as exc:
verbose_proxy_logger.error(
"Singulr API returned HTTP %s: %s",
exc.response.status_code,
str(exc),
)
if self.block_on_error:
raise GuardrailRaisedException(
guardrail_name=self.guardrail_name,
message=(f"Singulr API returned HTTP {exc.response.status_code}: {exc.response.text}"),
) from exc
return None
except httpx.TransportError as exc:
verbose_proxy_logger.error("Singulr API unreachable: %s", str(exc))
if self.block_on_error:
raise GuardrailRaisedException(
guardrail_name=self.guardrail_name,
message=f"Singulr API unreachable (block_on_error=True): {exc}",
) from exc
return None
except (ValueError, pydantic.ValidationError) as exc:
verbose_proxy_logger.error("Singulr API returned an invalid response: %s", str(exc))
if self.block_on_error:
raise GuardrailRaisedException(
guardrail_name=self.guardrail_name,
message=f"Singulr API returned an invalid response: {exc}",
) from exc
return None
@log_guardrail_information
async def apply_guardrail(
self,
inputs: GenericGuardrailAPIInputs,
request_data: dict,
input_type: str,
logging_obj: "LiteLLMLoggingObj | None" = None,
) -> GenericGuardrailAPIInputs:
payload = self._build_payload(request_data, inputs, input_type)
if not payload:
return inputs
result = await self._call_api(payload)
if result is None:
return inputs
verbose_proxy_logger.debug(
"Singulr: should_block=%s blocking_due_to=%s",
result.should_block,
result.blocking_due_to,
)
if result.should_block:
raise GuardrailRaisedException(
guardrail_name=self.guardrail_name,
message=f"Blocked by Singulr: {result.blocking_due_to or 'unknown'}",
)
return inputs

5
litellm/types/guardrails.py
View file

@ -53,6 +53,9 @@ from litellm.types.proxy.guardrails.guardrail_hooks.vigil_guard import (
from litellm.types.proxy.guardrails.guardrail_hooks.cisco_ai_defense import (
CiscoAIDefenseGuardrailConfigModel,
)
from litellm.types.proxy.guardrails.guardrail_hooks.singulr import (
SingulrGuardrailConfigModel,
)
from litellm.types.proxy.guardrails.guardrail_hooks.headroom import (
HeadroomGuardrailConfigModel,
)
@ -125,6 +128,7 @@ class SupportedGuardrailIntegrations(Enum):
RUBRIK = "rubrik"
VIGIL_GUARD = "vigil_guard"
REPELLOAI = "repelloai"
SINGULR = "singulr"
HEADROOM = "headroom"
COMPRESR = "compresr"
@ -924,6 +928,7 @@ class LitellmParams(
HiddenlayerGuardrailConfigModel,
QostodianNexusConfigModel,
VigilGuardGuardrailConfigModel,
SingulrGuardrailConfigModel,
):
guardrail: str = Field(description="The type of guardrail integration to use")
mode: Union[str, List[str], Mode] = Field(

63
litellm/types/proxy/guardrails/guardrail_hooks/singulr.py Normal file
View file

@ -0,0 +1,63 @@
from typing import Any, Optional
from pydantic import BaseModel, Field
from .base import GuardrailConfigModel
class SingulrGuardrailRequest(BaseModel):
model: Optional[str] = None
messages: Optional[list[dict[str, Any]]] = None
tools: Optional[list[dict[str, Any]]] = None
model_response: Optional[dict[str, Any]] = None
litellm_metadata: Optional[dict[str, Any]] = None
class SingulrGuardrailPayload(BaseModel):
litellm_call_id: Optional[str] = None
request_data: Optional[SingulrGuardrailRequest] = None
input_type: str
is_playground_request: Optional[bool] = None
playground_text: Optional[str] = None
class SingulrGuardrailResponse(BaseModel):
"""Response returned by the Singulr guardrail API."""
should_block: bool = False
blocking_due_to: Optional[str] = None
class SingulrGuardrailConfigModel(GuardrailConfigModel):
singulr_api_key: Optional[str] = Field(
default=None,
description="The Singulr API key. Generate API key from Singulr Platform.",
)
singulr_api_base: Optional[str] = Field(
default=None,
description="The Singulr API base URL. Get base URL from Singulr Platform.",
)
singulr_application_id: Optional[str] = Field(
default=None,
description="The Singulr application ID. Get application ID from Singulr Platform.",
)
singulr_guardrail_id: Optional[str] = Field(
default=None,
description="The Singulr Guardrail ID. Get guardrail ID from Singulr Platform.",
)
block_on_error: Optional[bool] = Field(
default=None,
description=(
"Whether to block requests when the Singulr Guardrails API is unavailable "
"or returns an error. If enabled, requests fail closed. "
"If disabled, requests continue without guardrail enforcement (fail open)."
),
)
@staticmethod
def ui_friendly_name() -> str:
return "Singulr"

550
tests/test_litellm/proxy/guardrails/guardrail_hooks/test_singulr.py Normal file
View file

@ -0,0 +1,550 @@
from unittest.mock import MagicMock, patch
import httpx
import pytest
from litellm.exceptions import GuardrailRaisedException
from litellm.proxy.guardrails.guardrail_hooks.singulr.singulr import SingulrGuardrail
from litellm.types.guardrails import GuardrailEventHooks
from litellm.types.proxy.guardrails.guardrail_hooks.singulr import (
SingulrGuardrailConfigModel,
)
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture
def singulr_guardrail():
"""Create a SingulrGuardrail instance with test credentials."""
return SingulrGuardrail(
singulr_api_base="https://api.test.singulr.ai",
singulr_api_key="test_token_1234",
singulr_guardrail_id="test_guardrail_id",
singulr_application_id="test_enforcement_entity",
guardrail_name="test-singulr",
event_hook="pre_call",
default_on=True,
)
def _make_response(body: dict) -> MagicMock:
"""Build a mock httpx response with the given JSON body."""
mock = MagicMock()
mock.json.return_value = body
mock.raise_for_status = MagicMock()
mock.status_code = 200
return mock
# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------
class TestSingulrConfiguration:
def test_init_with_explicit_credentials(self):
guardrail = SingulrGuardrail(
singulr_api_key="test_key",
singulr_api_base="https://custom.api.local",
singulr_guardrail_id="id123",
singulr_application_id="entity123",
guardrail_name="my-guardrail",
)
assert guardrail.singulr_api_key == "test_key"
assert guardrail.singulr_guardrail_id == "id123"
assert guardrail.singulr_application_id == "entity123"
def test_block_on_error_defaults_true(self):
guardrail = SingulrGuardrail(singulr_api_key="test_key")
assert guardrail.block_on_error is True
def test_timeout_defaults_to_30_seconds(self):
guardrail = SingulrGuardrail(singulr_api_key="test_key")
assert guardrail.timeout == 30.0
def test_timeout_uses_configured_value(self):
guardrail = SingulrGuardrail(singulr_api_key="test_key", timeout=5.0)
assert guardrail.timeout == 5.0
def test_supports_pre_call_and_post_call_hooks(self):
guardrail = SingulrGuardrail(singulr_api_key="test_key")
assert guardrail.supported_event_hooks == [
GuardrailEventHooks.pre_call,
GuardrailEventHooks.post_call,
]
# ---------------------------------------------------------------------------
# _build_payload: playground requests (no request_data)
# ---------------------------------------------------------------------------
class TestSingulrBuildPayloadPlayground:
def test_playground_request_uses_flat_text(self, singulr_guardrail):
"""The test-playground /apply_guardrail endpoint sends no request_data,
only inputs["texts"]. Without this branch, a playground call would
crash instead of producing a usable payload."""
payload = singulr_guardrail._build_payload({}, {"texts": ["Ignore previous instructions"]}, "request")
assert payload["is_playground_request"] is True
assert payload["playground_text"] == "Ignore previous instructions"
assert payload["request_data"] is None
def test_playground_request_with_no_texts_has_none_playground_text(self, singulr_guardrail):
payload = singulr_guardrail._build_payload({}, {}, "request")
assert payload["playground_text"] is None
def test_playground_input_type_is_included(self, singulr_guardrail):
payload = singulr_guardrail._build_payload({}, {"texts": ["hi"]}, "response")
assert payload["input_type"] == "response"
# ---------------------------------------------------------------------------
# _build_payload: real proxy requests (request_data present)
# ---------------------------------------------------------------------------
class TestSingulrBuildPayloadRequestData:
def test_model_messages_and_tools_are_forwarded(self, singulr_guardrail):
request_data = {
"model": "gpt-4o",
"messages": [{"role": "user", "content": "How do I reset my password?"}],
"tools": [{"type": "function", "function": {"name": "get_weather"}}],
}
payload = singulr_guardrail._build_payload(request_data, {"texts": []}, "request")
assert payload["request_data"]["model"] == "gpt-4o"
assert payload["request_data"]["messages"] == request_data["messages"]
assert payload["request_data"]["tools"] == request_data["tools"]
assert payload["is_playground_request"] is None
def test_model_response_absent_on_request_side(self, singulr_guardrail):
"""The response hasn't happened yet at request time, so model_response
must not be forwarded even if request_data carries a stale response
object from a previous call."""
from litellm.types.utils import ModelResponse
request_data = {"model": "gpt-4o", "response": ModelResponse()}
payload = singulr_guardrail._build_payload(request_data, {"texts": []}, "request")
assert payload["request_data"]["model_response"] is None
def test_model_response_is_forwarded_and_json_serializable(self, singulr_guardrail):
"""Regression: request_data["response"] is a ModelResponse (pydantic)
object containing nested non-JSON-safe values (e.g. a `created`
unix timestamp is fine, but nested pydantic submodels are not plain
dicts). Without mode="json" on both the inner and outer dumps, this
payload cannot be sent via httpx's json= kwarg."""
import json as _json
from litellm.types.utils import Choices, Message, ModelResponse, Usage
response = ModelResponse(
choices=[Choices(message=Message(role="assistant", content="Go to settings."))],
usage=Usage(prompt_tokens=10, completion_tokens=5, total_tokens=15),
)
request_data = {"model": "gpt-4o", "response": response}
payload = singulr_guardrail._build_payload(request_data, {"texts": ["Go to settings."]}, "response")
# Must not raise - this is what httpx's json= kwarg effectively does.
serialized = _json.dumps(payload)
assert "Go to settings." in serialized
assert payload["request_data"]["model_response"]["choices"][0]["message"]["content"] == "Go to settings."
def test_model_requested_tool_calls_are_forwarded_in_model_response(self, singulr_guardrail):
"""Tool calls the model requests arrive inside response.choices[].message.tool_calls.
They must survive the dump so Singulr can inspect what tools the
model is trying to invoke."""
from litellm.types.utils import Choices, Message, ModelResponse
response = ModelResponse(
choices=[
Choices(
message=Message(
role="assistant",
content=None,
tool_calls=[
{
"id": "call_1",
"type": "function",
"function": {"name": "get_current_time", "arguments": "{}"},
}
],
)
)
],
)
request_data = {"model": "gpt-4o", "response": response}
payload = singulr_guardrail._build_payload(request_data, {"texts": []}, "response")
tool_calls = payload["request_data"]["model_response"]["choices"][0]["message"]["tool_calls"]
assert tool_calls[0]["function"]["name"] == "get_current_time"
def test_litellm_metadata_is_forwarded(self, singulr_guardrail):
request_data = {"model": "gpt-4o", "litellm_metadata": {"user_api_key_hash": "abc123"}}
payload = singulr_guardrail._build_payload(request_data, {"texts": []}, "request")
assert payload["request_data"]["litellm_metadata"] == {"user_api_key_hash": "abc123"}
def test_internal_logging_object_is_not_forwarded(self, singulr_guardrail):
"""Regression: request_data can carry internal proxy objects (e.g. the
Logging instance) that aren't JSON-serializable at all. _build_payload
must only pull known request/response fields out of request_data,
not dump it wholesale, or this crashes on every real proxy call."""
import json as _json
class _NotSerializable:
pass
request_data = {
"model": "gpt-4o",
"messages": [{"role": "user", "content": "hi"}],
"litellm_logging_obj": _NotSerializable(),
}
payload = singulr_guardrail._build_payload(request_data, {"texts": ["hi"]}, "request")
# Must not raise.
_json.dumps(payload)
assert "litellm_logging_obj" not in payload["request_data"]
# ---------------------------------------------------------------------------
# Allow / block decisions
# ---------------------------------------------------------------------------
class TestSingulrAllowAction:
@pytest.mark.asyncio
async def test_allow_returns_inputs_unchanged(self, singulr_guardrail):
resp = _make_response({"should_block": False})
inputs = {"texts": ["How do I reset my password?"]}
with patch.object(singulr_guardrail.async_handler, "post", return_value=resp):
result = await singulr_guardrail.apply_guardrail(
inputs=inputs,
request_data={"model": "gpt-4o"},
input_type="request",
)
assert result is inputs
class TestSingulrBlockAction:
@pytest.mark.asyncio
async def test_block_raises_guardrail_exception(self, singulr_guardrail):
"""Regression: a should_block=True response must stop the request
instead of silently letting it through."""
resp = _make_response(
{
"should_block": True,
"blocking_due_to": "PII Information detected",
}
)
with patch.object(singulr_guardrail.async_handler, "post", return_value=resp):
with pytest.raises(GuardrailRaisedException) as exc_info:
await singulr_guardrail.apply_guardrail(
inputs={"texts": ["My SSN is 123-45-6789"]},
request_data={"model": "gpt-4o"},
input_type="request",
)
assert "PII Information detected" in str(exc_info.value)
@pytest.mark.asyncio
async def test_block_without_reason_uses_unknown_placeholder(self, singulr_guardrail):
resp = _make_response({"should_block": True})
with patch.object(singulr_guardrail.async_handler, "post", return_value=resp):
with pytest.raises(GuardrailRaisedException, match="unknown"):
await singulr_guardrail.apply_guardrail(
inputs={"texts": ["hi"]},
request_data={},
input_type="request",
)
# ---------------------------------------------------------------------------
# HTTP call wiring (endpoint, timeout, headers)
# ---------------------------------------------------------------------------
class TestSingulrRequestWiring:
@pytest.mark.asyncio
async def test_sends_configured_timeout(self):
"""litellm_params.timeout must reach the httpx call so operators can
tighten or loosen the latency budget instead of being stuck with a
hardcoded 30s regardless of configuration."""
guardrail = SingulrGuardrail(
singulr_api_key="test_key",
singulr_api_base="https://api.test.singulr.ai",
timeout=5.0,
)
resp = _make_response({"should_block": False})
with patch.object(guardrail.async_handler, "post", return_value=resp) as mock_post:
await guardrail.apply_guardrail(
inputs={"texts": ["test"]},
request_data={},
input_type="request",
)
assert mock_post.call_args.kwargs["timeout"] == 5.0
class TestSingulrBuildHeaders:
def test_content_type_always_present(self, singulr_guardrail):
assert singulr_guardrail._build_headers()["Content-Type"] == "application/json"
def test_all_optional_headers_included_when_set(self, singulr_guardrail):
headers = singulr_guardrail._build_headers()
assert headers["X-Singulr-Gateway-Token"] == "test_token_1234"
assert headers["X-Singulr-Enforcement-Entity-Id"] == "test_enforcement_entity"
assert headers["X-Singulr-Guardrail-Id"] == "test_guardrail_id"
def test_optional_headers_absent_when_unset(self):
guardrail = SingulrGuardrail(guardrail_name="bare")
headers = guardrail._build_headers()
assert "X-Singulr-Gateway-Token" not in headers
assert "X-Singulr-Enforcement-Entity-Id" not in headers
assert "X-Singulr-Guardrail-Id" not in headers
# ---------------------------------------------------------------------------
# Non-JSON / malformed response handling
# ---------------------------------------------------------------------------
class TestSingulrInvalidResponse:
@pytest.mark.asyncio
async def test_non_json_response_block_on_error_false_returns_inputs(self):
guardrail = SingulrGuardrail(
singulr_api_base="https://api.test.singulr.ai",
singulr_api_key="test_token_1234",
guardrail_name="test-singulr",
block_on_error=False,
)
mock_resp = MagicMock()
mock_resp.raise_for_status = MagicMock()
mock_resp.json.side_effect = ValueError("No JSON object could be decoded")
inputs = {"texts": ["test"]}
with patch.object(guardrail.async_handler, "post", return_value=mock_resp):
result = await guardrail.apply_guardrail(
inputs=inputs,
request_data={},
input_type="request",
)
assert result is inputs
@pytest.mark.asyncio
async def test_non_json_response_block_on_error_true_raises(self):
guardrail = SingulrGuardrail(
singulr_api_base="https://api.test.singulr.ai",
singulr_api_key="test_token_1234",
guardrail_name="test-singulr",
block_on_error=True,
)
mock_resp = MagicMock()
mock_resp.raise_for_status = MagicMock()
mock_resp.json.side_effect = ValueError("No JSON object could be decoded")
with patch.object(guardrail.async_handler, "post", return_value=mock_resp):
with pytest.raises(GuardrailRaisedException):
await guardrail.apply_guardrail(
inputs={"texts": ["test"]},
request_data={},
input_type="request",
)
@pytest.mark.asyncio
async def test_response_missing_expected_fields_block_on_error_true_raises(self):
"""Regression: a response body that fails SingulrGuardrailResponse
validation (e.g. should_block is a string, not a bool) must raise
GuardrailRaisedException instead of letting pydantic.ValidationError
propagate unhandled."""
guardrail = SingulrGuardrail(
singulr_api_base="https://api.test.singulr.ai",
singulr_api_key="test_token_1234",
guardrail_name="test-singulr",
block_on_error=True,
)
resp = _make_response({"should_block": "not-a-bool"})
with patch.object(guardrail.async_handler, "post", return_value=resp):
with pytest.raises(GuardrailRaisedException):
await guardrail.apply_guardrail(
inputs={"texts": ["test"]},
request_data={},
input_type="request",
)
# ---------------------------------------------------------------------------
# Transport error handling
# ---------------------------------------------------------------------------
class TestSingulrTransportError:
@pytest.mark.asyncio
async def test_remote_protocol_error_block_on_error_false_returns_inputs(self):
guardrail = SingulrGuardrail(
singulr_api_base="https://api.test.singulr.ai",
singulr_api_key="test_token_1234",
guardrail_name="test-singulr",
block_on_error=False,
)
inputs = {"texts": ["test"]}
with patch.object(
guardrail.async_handler,
"post",
side_effect=httpx.RemoteProtocolError("malformed HTTP response"),
):
result = await guardrail.apply_guardrail(
inputs=inputs,
request_data={},
input_type="request",
)
assert result is inputs
@pytest.mark.asyncio
async def test_remote_protocol_error_block_on_error_true_raises(self):
guardrail = SingulrGuardrail(
singulr_api_base="https://api.test.singulr.ai",
singulr_api_key="test_token_1234",
guardrail_name="test-singulr",
block_on_error=True,
)
with patch.object(
guardrail.async_handler,
"post",
side_effect=httpx.RemoteProtocolError("malformed HTTP response"),
):
with pytest.raises(GuardrailRaisedException):
await guardrail.apply_guardrail(
inputs={"texts": ["test"]},
request_data={},
input_type="request",
)
# ---------------------------------------------------------------------------
# HTTP status error handling
# ---------------------------------------------------------------------------
class TestSingulrHttpStatusError:
@pytest.mark.asyncio
async def test_http_error_message_names_status_code_not_unreachable(self):
guardrail = SingulrGuardrail(
singulr_api_base="https://api.test.singulr.ai",
singulr_api_key="test_token_1234",
guardrail_name="test-singulr",
block_on_error=True,
)
mock_response = MagicMock()
mock_response.status_code = 403
mock_response.text = "Forbidden"
exc = httpx.HTTPStatusError("403 Forbidden", request=MagicMock(), response=mock_response)
mock_response.raise_for_status.side_effect = exc
with patch.object(guardrail.async_handler, "post", return_value=mock_response):
with pytest.raises(GuardrailRaisedException) as exc_info:
await guardrail.apply_guardrail(
inputs={"texts": ["test"]},
request_data={},
input_type="request",
)
msg = str(exc_info.value)
assert "403" in msg
assert "unreachable" not in msg.lower()
@pytest.mark.asyncio
async def test_http_error_block_on_error_false_returns_inputs(self):
guardrail = SingulrGuardrail(
singulr_api_base="https://api.test.singulr.ai",
singulr_api_key="test_token_1234",
guardrail_name="test-singulr",
block_on_error=False,
)
mock_response = MagicMock()
mock_response.status_code = 500
mock_response.text = "Internal Server Error"
exc = httpx.HTTPStatusError("500", request=MagicMock(), response=mock_response)
mock_response.raise_for_status.side_effect = exc
inputs = {"texts": ["test"]}
with patch.object(guardrail.async_handler, "post", return_value=mock_response):
result = await guardrail.apply_guardrail(
inputs=inputs,
request_data={},
input_type="request",
)
assert result is inputs
# ---------------------------------------------------------------------------
# Config model
# ---------------------------------------------------------------------------
class TestSingulrConfigModel:
def test_ui_friendly_name(self):
assert SingulrGuardrailConfigModel.ui_friendly_name() == "Singulr"
# ---------------------------------------------------------------------------
# Initializer and registry
# ---------------------------------------------------------------------------
class TestSingulrInitializer:
def test_guardrail_initializer_registry_has_entry(self):
from litellm.proxy.guardrails.guardrail_hooks.singulr import (
initialize_guardrail,
)
assert callable(initialize_guardrail)
def test_initialize_guardrail_reads_singulr_prefixed_fields(self):
"""Regression: the UI config form (and YAML config) populate the
singulr_-prefixed fields declared on SingulrGuardrailConfigModel, not
the generic api_base/api_key fields. initialize_guardrail must read
those, or a UI-configured singulr_api_base is silently ignored and
the guardrail falls back to the localhost default."""
from litellm.proxy.guardrails.guardrail_hooks.singulr import (
initialize_guardrail,
)
from litellm.types.guardrails import Guardrail, LitellmParams
litellm_params = LitellmParams(
guardrail="singulr",
mode="pre_call",
singulr_api_base="https://configured.singulr.ai",
singulr_api_key="configured_key",
singulr_application_id="configured_app_id",
singulr_guardrail_id="configured_guardrail_id",
)
guardrail: Guardrail = {
"guardrail_name": "test-singulr",
"litellm_params": litellm_params,
}
cb = initialize_guardrail(litellm_params, guardrail)
assert cb.singulr_application_id == "configured_app_id"
assert cb.singulr_guardrail_id == "configured_guardrail_id"
def test_initialize_guardrail_wires_timeout(self):
"""BaseLitellmParams.timeout exists so operators can override the
per-request latency budget. initialize_guardrail must forward it to
SingulrGuardrail instead of leaving every deployment stuck on the
hardcoded default regardless of configuration."""
from litellm.proxy.guardrails.guardrail_hooks.singulr import (
initialize_guardrail,
)
from litellm.types.guardrails import Guardrail, LitellmParams
litellm_params = LitellmParams(
guardrail="singulr",
mode="pre_call",
singulr_api_key="configured_key",
timeout=12.5,
)
guardrail: Guardrail = {
"guardrail_name": "test-singulr",
"litellm_params": litellm_params,
}
cb = initialize_guardrail(litellm_params, guardrail)
assert cb.timeout == 12.5