From c4064125f569112da7d104e0e78f244d67ead34a Mon Sep 17 00:00:00 2001 From: Sudhi Seshachala Date: Mon, 24 Aug 2026 15:47:10 -0500 Subject: [PATCH] feat(guardrails): add ConductGuard integration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds Conduct Guard as a first-class LiteLLM guardrail. Point any LiteLLM proxy at Conduct and every LLM call routed through it is policy-checked before the upstream request goes out — block, warn, audit, or trigger a human-in-the-loop approval, with the same signed configuration + hash-chained audit log Conduct exposes on its native enforcement surfaces. - litellm/types/guardrails.py: add CONDUCT to SupportedGuardrailIntegrations. - litellm/proxy/guardrails/guardrail_hooks/conduct/__init__.py: registration via guardrail_initializer_registry and guardrail_class_registry, picked up by the auto-discovery in guardrail_registry.py. - litellm/proxy/guardrails/guardrail_hooks/conduct/conduct.py: the adapter. CustomGuardrail subclass, async_pre_call_hook, response envelope parser for the five Conduct verdicts (ok / advisory / WARNING / BLOCKED / PENDING approval), fail-mode logic, session-ID resolution chain (litellm_metadata.trace_id → X-Conduct-Session-Id → hash fallback). - tests/test_litellm/proxy/guardrails/test_conduct_guardrail.py: envelope parsing, pre-call allow/block/approval, config precedence, missing-token construction error. ```yaml guardrails: - guardrail_name: conduct-guard litellm_params: guardrail: conduct mode: pre_call api_base: https://api.conductai.ai # optional, default api_key: os.environ/CONDUCT_AGENT_TOKEN # cond_agt_* token fail_mode: fail_closed # or fail_open tool_name: llm_call # scoped tool_name ``` A standalone PyPI package `conduct-litellm-guard` shipped ahead of this PR for teams pinned to older LiteLLM versions. Once this integration merges, the standalone README will point at the native support as the preferred path. - PyPI: https://pypi.org/project/conduct-litellm-guard/ - Product: https://conductai.ai/guard Contact: sudhi@b2bsphere.com --- .../guardrail_hooks/conduct/__init__.py | 37 +++ .../guardrail_hooks/conduct/conduct.py | 311 ++++++++++++++++++ litellm/types/guardrails.py | 1 + .../guardrails/test_conduct_guardrail.py | 122 +++++++ 4 files changed, 471 insertions(+) create mode 100644 litellm/proxy/guardrails/guardrail_hooks/conduct/__init__.py create mode 100644 litellm/proxy/guardrails/guardrail_hooks/conduct/conduct.py create mode 100644 tests/test_litellm/proxy/guardrails/test_conduct_guardrail.py diff --git a/litellm/proxy/guardrails/guardrail_hooks/conduct/__init__.py b/litellm/proxy/guardrails/guardrail_hooks/conduct/__init__.py new file mode 100644 index 00000000000..868d0f53143 --- /dev/null +++ b/litellm/proxy/guardrails/guardrail_hooks/conduct/__init__.py @@ -0,0 +1,37 @@ +from typing import TYPE_CHECKING, Final + +from litellm.types.guardrails import SupportedGuardrailIntegrations + +from .conduct import ConductGuardrail + +if TYPE_CHECKING: + from litellm.types.guardrails import Guardrail, LitellmParams + + +def initialize_guardrail(litellm_params: "LitellmParams", guardrail: "Guardrail"): + import litellm + + _conduct_callback: Final = ConductGuardrail( + api_base=getattr(litellm_params, "api_base", None), + api_key=getattr(litellm_params, "api_key", None), + workspace_id=getattr(litellm_params, "workspace_id", None), + fail_mode=getattr(litellm_params, "fail_mode", "fail_closed"), + tool_name=getattr(litellm_params, "tool_name", "llm_call"), + timeout=getattr(litellm_params, "timeout", 8.0), + guardrail_name=guardrail.get("guardrail_name", ""), + event_hook=litellm_params.mode, + default_on=litellm_params.default_on, + ) + litellm.logging_callback_manager.add_litellm_callback(_conduct_callback) + + return _conduct_callback + + +guardrail_initializer_registry: Final = { + SupportedGuardrailIntegrations.CONDUCT.value: initialize_guardrail, +} + + +guardrail_class_registry: Final = { + SupportedGuardrailIntegrations.CONDUCT.value: ConductGuardrail, +} diff --git a/litellm/proxy/guardrails/guardrail_hooks/conduct/conduct.py b/litellm/proxy/guardrails/guardrail_hooks/conduct/conduct.py new file mode 100644 index 00000000000..6783a46b3bc --- /dev/null +++ b/litellm/proxy/guardrails/guardrail_hooks/conduct/conduct.py @@ -0,0 +1,311 @@ +# +-------------------------------------------------------------+ +# +# Use Conduct Guard for your LLM calls +# +# Runtime policy enforcement — block / warn / audit / approval +# Signed configuration + hash-chained audit + 20+ compliance packs +# Docs: https://conductai.ai/guard +# +# +-------------------------------------------------------------+ +from __future__ import annotations + +import hashlib +import os +import uuid +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any, Final, Literal + +from litellm._logging import verbose_proxy_logger +from litellm.integrations.custom_guardrail import ( + CustomGuardrail, + log_guardrail_information, +) +from litellm.llms.custom_httpx.http_handler import ( + get_async_httpx_client, + httpxSpecialProvider, +) +from litellm.proxy._types import UserAPIKeyAuth +from litellm.types.guardrails import GuardrailEventHooks + +GUARDRAIL_NAME: Final = "conduct" + +if TYPE_CHECKING: + from litellm.types.proxy.guardrails.guardrail_hooks.base import GuardrailConfigModel + + +Verdict = Literal["allow", "advisory", "warning", "block", "approval", "unknown"] +FailMode = Literal["fail_open", "fail_closed"] + + +@dataclass(frozen=True) +class GuardDecision: + """Structured view of what ``guard_check`` returned. The raw text is + kept so audit / logging surfaces can quote it verbatim.""" + + verdict: Verdict + raw: str + rule_id: str | None = None + message: str | None = None + + @classmethod + def parse(cls, text: str) -> "GuardDecision": + """Map the ``guard_check`` string envelope to a verdict. + + Response contract from Conduct: + * ``"ok"`` or empty → allow silently + * ``"advisory: ..."`` → allow but log + * ``"WARNING — ..."`` → allow but surface + * ``"BLOCKED — ..."`` → hard block + * ``"PENDING approval — ..."`` → HITL — treat as block + """ + stripped = (text or "").strip() + if not stripped or stripped.lower().startswith("ok"): + return cls(verdict="allow", raw=stripped) + if stripped.startswith("BLOCKED"): + return cls( + verdict="block", + raw=stripped, + rule_id=_extract_rule_id(stripped), + message=_strip_prefix(stripped, "BLOCKED"), + ) + if stripped.startswith("PENDING approval"): + return cls( + verdict="approval", + raw=stripped, + rule_id=_extract_rule_id(stripped), + message=_strip_prefix(stripped, "PENDING approval"), + ) + if stripped.startswith("WARNING"): + return cls( + verdict="warning", + raw=stripped, + rule_id=_extract_rule_id(stripped), + message=_strip_prefix(stripped, "WARNING"), + ) + if stripped.startswith("advisory"): + return cls( + verdict="advisory", + raw=stripped, + rule_id=_extract_rule_id(stripped), + message=_strip_prefix(stripped, "advisory"), + ) + return cls(verdict="unknown", raw=stripped) + + +def _strip_prefix(text: str, prefix: str) -> str | None: + remainder = text[len(prefix):].strip() + return remainder.lstrip(":—- ").strip() or None + + +def _extract_rule_id(text: str) -> str | None: + marker = "[rule:" + idx = text.find(marker) + if idx < 0: + return None + tail = text[idx + len(marker):] + end = tail.find("]") + return tail[:end].strip() if end >= 0 else None + + +class ConductGuardrailBlocked(Exception): + """Raised inside the pre-call hook to abort a LiteLLM request. LiteLLM + surfaces the message to the caller.""" + + def __init__(self, decision: GuardDecision) -> None: + self.decision = decision + super().__init__(decision.message or decision.raw or "Blocked by Conduct Guard") + + +class ConductGuardrail(CustomGuardrail): + """Conduct Guard as a LiteLLM ``CustomGuardrail``. + + Reads config from LiteLLM's guardrail block. Every pre-call hook + invocation calls Conduct's ``guard_check`` MCP tool using the + supplied agent token. On block, raises so the LiteLLM proxy returns + an error to the caller instead of forwarding to the model.""" + + @classmethod + def get_supported_event_hooks(cls) -> list[GuardrailEventHooks]: + return [ + GuardrailEventHooks.pre_call, + GuardrailEventHooks.during_call, + ] + + def __init__( + self, + *, + api_base: str | None = None, + api_key: str | None = None, + workspace_id: str | None = None, + fail_mode: FailMode = "fail_closed", + tool_name: str = "llm_call", + timeout: float = 8.0, + **kwargs: Any, + ) -> None: + kwargs.setdefault("supported_event_hooks", list(self.get_supported_event_hooks())) + super().__init__(**kwargs) + + self._api_url = ( + api_base or os.environ.get("CONDUCT_API_URL", "https://api.conductai.ai") + ).rstrip("/") + token = api_key or os.environ.get("CONDUCT_AGENT_TOKEN") + if not token: + raise ValueError( + "ConductGuardrail: agent token is required. Set CONDUCT_AGENT_TOKEN " + "in the environment or pass api_key in the guardrail config." + ) + self._agent_token = token + self._workspace_id = workspace_id or os.environ.get("CONDUCT_WORKSPACE_ID") + self._fail_mode: FailMode = fail_mode + self._tool_name = tool_name + self._timeout = timeout + self._async_handler = get_async_httpx_client(llm_provider=httpxSpecialProvider.GuardrailCallback) + + # ── LiteLLM contract ─────────────────────────────────────────────── + + @log_guardrail_information + async def async_pre_call_hook( + self, + user_api_key_dict: UserAPIKeyAuth, + cache: Any, + data: dict[str, Any], + call_type: Literal[ + "completion", + "text_completion", + "embeddings", + "image_generation", + "moderation", + "audio_transcription", + ], + ) -> dict[str, Any] | None: + decision = await self._check(data=data, call_type=call_type) + + if decision.verdict in ("block", "approval"): + raise ConductGuardrailBlocked(decision) + + data.setdefault("metadata", {}).setdefault("conduct_guard", {}).update( + {"verdict": decision.verdict, "rule_id": decision.rule_id} + ) + return data + + # ── Guard check ───────────────────────────────────────────────── + + async def _check(self, *, data: dict[str, Any], call_type: str) -> GuardDecision: + tool_input = _build_tool_input(data, call_type) + session_id = _extract_session_id(data) + prompt = _extract_prompt_text(data) + + arguments: dict[str, Any] = { + "tool_name": self._tool_name, + "tool_input": tool_input, + } + if prompt is not None: + arguments["prompt"] = prompt + + payload = { + "jsonrpc": "2.0", + "id": str(uuid.uuid4()), + "method": "tools/call", + "params": {"name": "guard_check", "arguments": arguments}, + } + + headers = { + "Authorization": f"Bearer {self._agent_token}", + "Content-Type": "application/json", + "User-Agent": "litellm-conduct-guardrail/1.0", + "X-Claude-Surface": "litellm", + } + if self._workspace_id: + headers["X-Workspace-Id"] = self._workspace_id + if session_id: + headers["X-Conduct-Session-Id"] = session_id + + try: + response = await self._async_handler.post( + f"{self._api_url}/guard/mcp", + json=payload, + headers=headers, + timeout=self._timeout, + ) + response.raise_for_status() + body = response.json() + + if "error" in body: + err = body["error"] + verbose_proxy_logger.warning( + "conduct_guard: eval error %s — applying %s", err, self._fail_mode + ) + if self._fail_mode == "fail_closed": + return GuardDecision( + verdict="block", + raw=str(err), + message="Conduct Guard policy-eval error (fail_closed).", + ) + return GuardDecision(verdict="allow", raw="fail_open") + + result = body.get("result") or {} + for item in result.get("content", []) or []: + if item.get("type") == "text": + return GuardDecision.parse(item.get("text", "")) + return GuardDecision(verdict="allow", raw="") + except Exception as e: + verbose_proxy_logger.warning( + "conduct_guard: transport error %s — applying %s", e, self._fail_mode + ) + if self._fail_mode == "fail_closed": + return GuardDecision( + verdict="block", + raw=str(e), + message="Conduct Guard is unreachable (fail_closed).", + ) + return GuardDecision(verdict="allow", raw="fail_open") + + +# ── Helpers ───────────────────────────────────────────────────────── + + +def _extract_session_id(data: dict[str, Any]) -> str | None: + metadata = data.get("litellm_metadata") or data.get("metadata") or {} + for key in ("trace_id", "X-Conduct-Session-Id", "conduct_session_id"): + val = metadata.get(key) + if val: + return str(val) + + user = data.get("user") or metadata.get("user") or "" + first_msg = "" + for m in data.get("messages") or []: + if isinstance(m, dict) and m.get("role") == "user": + first_msg = str(m.get("content", ""))[:512] + break + if not user and not first_msg: + return None + digest = hashlib.sha256((user + "|" + first_msg).encode("utf-8")).hexdigest() + return f"litellm-{digest[:16]}" + + +def _extract_prompt_text(data: dict[str, Any]) -> str | None: + for m in reversed(data.get("messages") or []): + if isinstance(m, dict) and m.get("role") == "user": + content = m.get("content") + if isinstance(content, str): + return content[:4000] + if isinstance(content, list): + parts = [ + p.get("text", "") for p in content + if isinstance(p, dict) and p.get("type") == "text" + ] + return " ".join(parts)[:4000] or None + return None + + +def _build_tool_input(data: dict[str, Any], call_type: str) -> dict[str, Any]: + messages = data.get("messages") or [] + return { + "model": data.get("model"), + "call_type": call_type, + "message_count": len(messages), + "temperature": data.get("temperature"), + "max_tokens": data.get("max_tokens"), + "stream": bool(data.get("stream")), + "content": _extract_prompt_text(data) or "", + } diff --git a/litellm/types/guardrails.py b/litellm/types/guardrails.py index 02dee40f2a3..69cb88bfa2f 100644 --- a/litellm/types/guardrails.py +++ b/litellm/types/guardrails.py @@ -137,6 +137,7 @@ class SupportedGuardrailIntegrations(Enum): COMPRESR = "compresr" STRAIKER = "straiker" ALICE = "alice" + CONDUCT = "conduct" class Role(Enum): diff --git a/tests/test_litellm/proxy/guardrails/test_conduct_guardrail.py b/tests/test_litellm/proxy/guardrails/test_conduct_guardrail.py new file mode 100644 index 00000000000..b776b013d72 --- /dev/null +++ b/tests/test_litellm/proxy/guardrails/test_conduct_guardrail.py @@ -0,0 +1,122 @@ +"""Unit tests for the Conduct guardrail. + +Mocked transport — no real network. Verifies the response-envelope +parser, pre-call hook behavior (allow / block / approval), fail-mode +handling, and session-ID resolution chain. +""" +from __future__ import annotations + +import os +from unittest.mock import AsyncMock + +import pytest + +from litellm.proxy.guardrails.guardrail_hooks.conduct.conduct import ( + ConductGuardrail, + ConductGuardrailBlocked, + GuardDecision, +) + + +# ── Decision parsing ──────────────────────────────────────────────────── + + +class TestGuardDecisionParse: + @pytest.mark.parametrize("raw", ["ok", "OK", "", " ok "]) + def test_ok_variants_are_allow(self, raw: str) -> None: + assert GuardDecision.parse(raw).verdict == "allow" + + def test_blocked_extracts_rule_id(self) -> None: + d = GuardDecision.parse( + "BLOCKED — command touches /etc/passwd [rule: no-etc-passwd]" + ) + assert d.verdict == "block" + assert d.rule_id == "no-etc-passwd" + + def test_pending_approval_treated_as_block(self) -> None: + d = GuardDecision.parse( + "PENDING approval — HITL required [rule: prod-deploy-gate]" + ) + assert d.verdict == "approval" + assert d.rule_id == "prod-deploy-gate" + + def test_warning_is_warning(self) -> None: + d = GuardDecision.parse("WARNING — high-risk model [rule: model-tier]") + assert d.verdict == "warning" + + def test_advisory_is_advisory(self) -> None: + d = GuardDecision.parse("advisory: policy eval error: boom") + assert d.verdict == "advisory" + + def test_unknown_prefix_marked_unknown(self) -> None: + assert GuardDecision.parse("wat").verdict == "unknown" + + +# ── Pre-call hook ────────────────────────────────────────────────────── + + +@pytest.fixture(autouse=True) +def _agent_token_env(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("CONDUCT_AGENT_TOKEN", "cond_agt_test_placeholder") + + +def _guard() -> ConductGuardrail: + return ConductGuardrail() + + +@pytest.mark.asyncio +class TestPreCallHook: + async def test_allow_returns_data_with_metadata_tag(self) -> None: + g = _guard() + g._check = AsyncMock(return_value=GuardDecision(verdict="allow", raw="ok")) + data = {"model": "gpt-4o", "messages": [{"role": "user", "content": "hi"}]} + result = await g.async_pre_call_hook(None, None, data, "completion") + assert result is data + assert result["metadata"]["conduct_guard"]["verdict"] == "allow" + + async def test_block_raises(self) -> None: + g = _guard() + g._check = AsyncMock( + return_value=GuardDecision( + verdict="block", + raw="BLOCKED — no secrets [rule: no-prod-secrets]", + rule_id="no-prod-secrets", + ) + ) + with pytest.raises(ConductGuardrailBlocked) as exc: + await g.async_pre_call_hook(None, None, {}, "completion") + assert exc.value.decision.rule_id == "no-prod-secrets" + + async def test_pending_approval_also_raises(self) -> None: + g = _guard() + g._check = AsyncMock( + return_value=GuardDecision(verdict="approval", raw="PENDING approval — review") + ) + with pytest.raises(ConductGuardrailBlocked): + await g.async_pre_call_hook(None, None, {}, "completion") + + +class TestConfig: + def test_missing_token_raises_at_construction( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.delenv("CONDUCT_AGENT_TOKEN", raising=False) + with pytest.raises(ValueError, match="agent token"): + ConductGuardrail() + + def test_config_api_key_wins_over_env( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("CONDUCT_AGENT_TOKEN", "env-token") + g = ConductGuardrail(api_key="config-token") + assert g._agent_token == "config-token" + + def test_config_api_base_wins_over_default(self) -> None: + g = ConductGuardrail(api_base="https://conduct.example.com/") + assert g._api_url == "https://conduct.example.com" + + +if __name__ == "__main__": + import subprocess + import sys + raise SystemExit(subprocess.call([sys.executable, "-m", "pytest", __file__, "-v"]))