diff --git a/docs/my-website/docs/proxy/guardrails/akto.md b/docs/my-website/docs/proxy/guardrails/akto.md new file mode 100644 index 0000000000..67ae741d11 --- /dev/null +++ b/docs/my-website/docs/proxy/guardrails/akto.md @@ -0,0 +1,139 @@ +# Akto + +## Overview +[Akto](https://www.akto.io/) provides API security guardrails and data ingestion for LLM traffic. + +Akto now uses a **two-entry guardrail pattern** in LiteLLM: +- `akto-validate` (`pre_call`) for request validation +- `akto-ingest` (`post_call`) for request/response ingestion + +There is no `on_flagged` setting anymore. + +Use these as two separate guardrails in `config.yaml`: +- `guardrail_name: "akto-validate"` +- `guardrail_name: "akto-ingest"` + +## 1. Get Your Akto Credentials + +Set up the Akto Guardrail API Service and grab: +- `AKTO_GUARDRAIL_API_BASE` — your Guardrail API Base URL +- `AKTO_API_KEY` — your API key + +## 2. Configure in `config.yaml` + +### Block + Ingest (recommended) + +Use both entries below. This gives you: +- pre-call block decision +- post-call ingestion for allowed traffic + +Keep these as two separate entries (`akto-validate` and `akto-ingest`). + +```yaml +guardrails: + - guardrail_name: "akto-validate" + litellm_params: + guardrail: akto + mode: pre_call + akto_base_url: os.environ/AKTO_GUARDRAIL_API_BASE + akto_api_key: os.environ/AKTO_API_KEY + default_on: true + unreachable_fallback: fail_closed # optional: fail_open | fail_closed (default: fail_closed) + guardrail_timeout: 5 # optional, default: 5 + akto_account_id: "1000000" # optional, env fallback: AKTO_ACCOUNT_ID + akto_vxlan_id: "0" # optional, env fallback: AKTO_VXLAN_ID + + - guardrail_name: "akto-ingest" + litellm_params: + guardrail: akto + mode: post_call + akto_base_url: os.environ/AKTO_GUARDRAIL_API_BASE + akto_api_key: os.environ/AKTO_API_KEY + default_on: true +``` + +### Monitor-only mode + +If you only want logging/ingestion and no blocking, keep only `akto-ingest`. + +```yaml +guardrails: + - guardrail_name: "akto-ingest" + litellm_params: + guardrail: akto + mode: post_call + akto_base_url: os.environ/AKTO_GUARDRAIL_API_BASE + akto_api_key: os.environ/AKTO_API_KEY + default_on: true +``` + +## 3. Test It + +```shell +curl -i http://localhost:4000/v1/chat/completions \ + -H "Content-Type: application/json" \ + -H "Authorization: Bearer " \ + -d '{ + "model": "gpt-3.5-turbo", + "messages": [ + {"role": "user", "content": "Hello, how are you?"} + ] + }' +``` + +If a request gets blocked: + +```json +{ + "error": { + "message": "Prompt injection detected", + "type": "None", + "param": "None", + "code": "403" + } +} +``` + +## 4. How It Works + +**Block + Ingest mode:** +``` +Request → LiteLLM → Akto guardrail check + → Allowed → forward to LLM → ingest response + → Blocked → ingest blocked marker → 403 error +``` + +**Monitor-only mode:** +``` +Request → LiteLLM → forward to LLM → get response + → Send to Akto (guardrails + ingest) → log only +``` + +## 5. Event behavior + +| Entry | LiteLLM hook | Akto call behavior | +|------|---|---| +| `akto-validate` | `pre_call` | Awaited call with `guardrails=true`, `ingest_data=false` | +| `akto-ingest` | `post_call` | Fire-and-forget call with `guardrails=true`, `ingest_data=true` | + +When blocked in `pre_call`, LiteLLM sends one fire-and-forget ingest payload with blocked metadata and returns `403`. + +## 6. Parameters + +| Parameter | Env Variable | Default | Description | +|-----------|-------------|---------|-------------| +| `akto_base_url` | `AKTO_GUARDRAIL_API_BASE` | *required* | Akto Guardrail API Base URL | +| `akto_api_key` | `AKTO_API_KEY` | *required* | API key (sent as `Authorization` header) | +| `akto_account_id` | `AKTO_ACCOUNT_ID` | `1000000` | Akto account id included in payload | +| `akto_vxlan_id` | `AKTO_VXLAN_ID` | `0` | Akto vxlan id included in payload | +| `unreachable_fallback` | — | `fail_closed` | `fail_open` or `fail_closed` | +| `guardrail_timeout` | — | `5` | Timeout in seconds | +| `default_on` | — | `true` (recommended) | Enables the guardrail entry by default | + +## 7. Error Handling + +| Scenario | `fail_closed` (default) | `fail_open` | +|----------|------------------------|-------------| +| Akto unreachable | ❌ Blocked (503) | ✅ Passes through | +| Akto returns error | ❌ Blocked (503) | ✅ Passes through | +| Guardrail says no | ❌ Blocked (403) | ❌ Blocked (403) | diff --git a/docs/my-website/sidebars.js b/docs/my-website/sidebars.js index 1362745a91..e53891d633 100644 --- a/docs/my-website/sidebars.js +++ b/docs/my-website/sidebars.js @@ -52,6 +52,7 @@ const sidebars = { label: "Providers", items: [ ...[ + "proxy/guardrails/akto", "proxy/guardrails/qualifire", "proxy/guardrails/aim_security", "proxy/guardrails/onyx_security", diff --git a/litellm/proxy/guardrails/guardrail_hooks/akto/__init__.py b/litellm/proxy/guardrails/guardrail_hooks/akto/__init__.py new file mode 100644 index 0000000000..4ae2675540 --- /dev/null +++ b/litellm/proxy/guardrails/guardrail_hooks/akto/__init__.py @@ -0,0 +1,37 @@ +from typing import TYPE_CHECKING + +from litellm.types.guardrails import SupportedGuardrailIntegrations + +from .akto import AktoGuardrail + + +if TYPE_CHECKING: + from litellm.types.guardrails import Guardrail, LitellmParams + + +def initialize_guardrail(litellm_params: "LitellmParams", guardrail: "Guardrail"): + import litellm + + _akto_callback = AktoGuardrail( + akto_base_url=getattr(litellm_params, "akto_base_url", None), + akto_api_key=getattr(litellm_params, "akto_api_key", None), + akto_account_id=getattr(litellm_params, "akto_account_id", None), + akto_vxlan_id=getattr(litellm_params, "akto_vxlan_id", None), + unreachable_fallback=getattr(litellm_params, "unreachable_fallback", "fail_closed"), + guardrail_timeout=getattr(litellm_params, "guardrail_timeout", None), + guardrail_name=guardrail.get("guardrail_name", ""), + event_hook=litellm_params.mode, + default_on=litellm_params.default_on, + ) + + litellm.logging_callback_manager.add_litellm_callback(_akto_callback) + return _akto_callback + + +guardrail_initializer_registry = { + SupportedGuardrailIntegrations.AKTO.value: initialize_guardrail, +} + +guardrail_class_registry = { + SupportedGuardrailIntegrations.AKTO.value: AktoGuardrail, +} diff --git a/litellm/proxy/guardrails/guardrail_hooks/akto/akto.py b/litellm/proxy/guardrails/guardrail_hooks/akto/akto.py new file mode 100644 index 0000000000..be9c9cb1be --- /dev/null +++ b/litellm/proxy/guardrails/guardrail_hooks/akto/akto.py @@ -0,0 +1,456 @@ +"""Akto guardrail integration for LiteLLM proxy. + +Uses a two-config-entry pattern: + - akto-validate (pre_call): Checks request against Akto guardrails, blocks if flagged. + - akto-ingest (post_call): Sends request+response to Akto for data ingestion. + +For monitor-only mode, enable only akto-ingest without akto-validate. +""" + +import asyncio +import json +import os +from datetime import datetime +from typing import TYPE_CHECKING, Any, Dict, Literal, Optional, Tuple, Type + +from fastapi import HTTPException + +import httpx + +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.types.guardrails import GuardrailEventHooks +from litellm.types.utils import GenericGuardrailAPIInputs + +if TYPE_CHECKING: + from litellm.types.proxy.guardrails.guardrail_hooks.base import GuardrailConfigModel + +HTTP_PROXY_PATH = "/api/http-proxy" +AKTO_CONNECTOR_NAME = "litellm" +DEFAULT_GUARDRAIL_TIMEOUT = 5 + + +class AktoGuardrail(CustomGuardrail): + """LiteLLM guardrail hook that validates and ingests LLM traffic via the Akto API.""" + + # Maps event_hook to the input_type it should handle; mismatches are no-ops + HOOK_TO_INPUT = {"pre_call": "request", "post_call": "response"} + + @staticmethod + def get_config_model() -> Type["GuardrailConfigModel"]: + """Return the Pydantic config model for YAML-based initialization.""" + from litellm.types.proxy.guardrails.guardrail_hooks.akto import ( + AktoConfigModel, + ) + + return AktoConfigModel + + def __init__( + self, + akto_base_url: Optional[str] = None, + akto_api_key: Optional[str] = None, + akto_account_id: Optional[str] = None, + akto_vxlan_id: Optional[str] = None, + unreachable_fallback: Literal["fail_closed", "fail_open"] = "fail_closed", + guardrail_timeout: Optional[int] = None, + **kwargs: Any, + ) -> None: + """Initialize the Akto guardrail. + + Args: + akto_base_url: Akto API base URL. Falls back to AKTO_GUARDRAIL_API_BASE env var. + akto_api_key: Akto API key. Falls back to AKTO_API_KEY env var. + akto_account_id: Akto account ID. Falls back to AKTO_ACCOUNT_ID env var, then "1000000". + akto_vxlan_id: Akto VXLAN ID. Falls back to AKTO_VXLAN_ID env var, then "0". + unreachable_fallback: Behavior when Akto is unreachable — block or allow. + guardrail_timeout: HTTP timeout in seconds for Akto API calls. + """ + self.async_handler = get_async_httpx_client( + llm_provider=httpxSpecialProvider.GuardrailCallback, + ) + self.background_tasks: set = set() + + self.akto_base_url = (akto_base_url or os.environ.get("AKTO_GUARDRAIL_API_BASE", "")).rstrip("/") + if not self.akto_base_url: + raise ValueError("akto_base_url is required. Set AKTO_GUARDRAIL_API_BASE or pass it in litellm_params.") + + self.akto_api_key = akto_api_key or os.environ.get("AKTO_API_KEY", "") + if not self.akto_api_key: + raise ValueError("akto_api_key is required. Set AKTO_API_KEY or pass it in litellm_params.") + + self.unreachable_fallback: Literal["fail_closed", "fail_open"] = unreachable_fallback + self.guardrail_timeout = guardrail_timeout or DEFAULT_GUARDRAIL_TIMEOUT + self.akto_account_id = akto_account_id or os.environ.get("AKTO_ACCOUNT_ID", "1000000") + self.akto_vxlan_id = akto_vxlan_id or os.environ.get("AKTO_VXLAN_ID", "0") + + kwargs["supported_event_hooks"] = [ + GuardrailEventHooks.pre_call, + GuardrailEventHooks.post_call, + ] + super().__init__(**kwargs) + + verbose_proxy_logger.debug( + "Akto guardrail initialized: base_url=%s fallback=%s", + self.akto_base_url, + self.unreachable_fallback, + ) + + @staticmethod + def resolve_metadata_value(request_data: Optional[dict], key: str) -> Optional[str]: + """Look up a metadata value from litellm_metadata or metadata dicts.""" + if request_data is None: + return None + for dict_key in ("litellm_metadata", "metadata"): + container = request_data.get(dict_key) or {} + if isinstance(container, dict) and container: + value = container.get(key) + if value is not None: + return str(value).strip() + return None + + @staticmethod + def extract_request_path(request_data: dict) -> str: + """Extract the API route from request metadata, defaulting to /v1/chat/completions.""" + metadata = request_data.get("metadata") or {} + if not isinstance(metadata, dict): + metadata = {} + route = metadata.get("user_api_key_request_route") + return route if route else "/v1/chat/completions" + + def prepare_headers(self) -> Dict[str, str]: + """Build HTTP headers for the Akto API call.""" + return { + "content-type": "application/json", + "Authorization": self.akto_api_key, + } + + @staticmethod + def build_query_params(*, guardrails: bool, ingest_data: bool) -> Dict[str, str]: + """Build query params that control Akto backend behavior (guardrail check and/or data ingestion).""" + params: Dict[str, str] = {"akto_connector": AKTO_CONNECTOR_NAME} + if guardrails: + params["guardrails"] = "true" + if ingest_data: + params["ingest_data"] = "true" + return params + + @staticmethod + def build_request_headers(request_data: dict) -> Dict[str, str]: + """Build the requestHeaders field from proxy request headers.""" + headers: Dict[str, str] = {"content-type": "application/json"} + proxy_req = request_data.get("proxy_server_request", {}) + if not isinstance(proxy_req, dict): + return headers + proxy_req_headers = proxy_req.get("headers") + if isinstance(proxy_req_headers, dict): + for key, val in proxy_req_headers.items(): + if key and val: + headers[str(key).lower()] = str(val) + return headers + + @staticmethod + def build_request_body( + inputs: GenericGuardrailAPIInputs, + request_data: Optional[dict] = None, + ) -> Dict[str, Any]: + """Build the LLM request body from guardrail inputs (messages, model, tools).""" + model = inputs.get("model", "") or "" + body: Dict[str, Any] = {"model": model} + + structured = inputs.get("structured_messages") + if structured: + body["messages"] = structured + elif request_data is not None and request_data.get("messages"): + body["messages"] = request_data["messages"] + if request_data.get("model"): + body["model"] = request_data["model"] + else: + texts = inputs.get("texts", []) + body["messages"] = [{"role": "user", "content": t} for t in texts] if texts else [] + + tools = inputs.get("tools") + if tools: + body["tools"] = tools + elif request_data is not None and request_data.get("tools"): + body["tools"] = request_data["tools"] + + tool_calls = inputs.get("tool_calls") + if tool_calls: + body["tool_calls"] = tool_calls + + return body + + @staticmethod + def build_response_body( + inputs: GenericGuardrailAPIInputs, + request_data: Optional[dict] = None, + ) -> Dict[str, Any]: + """Build the LLM response body, preferring the actual model response if available.""" + model_response = request_data.get("response") if request_data else None + if model_response is not None and hasattr(model_response, "model_dump"): + return model_response.model_dump() + + texts = inputs.get("texts", []) + if texts: + return {"choices": [{"message": {"content": t, "role": "assistant"}} for t in texts]} + return {} + + @staticmethod + def build_tag_metadata(request_data: dict) -> Dict[str, str]: + """Build tag/metadata dict with user_id and team_id for Akto tracking.""" + tag: Dict[str, str] = {"gen-ai": "Gen AI"} + user_id = AktoGuardrail.resolve_metadata_value(request_data, "user_api_key_user_id") + team_id = AktoGuardrail.resolve_metadata_value(request_data, "user_api_key_team_id") + if user_id: + tag["user_id"] = user_id + if team_id: + tag["team_id"] = team_id + return tag + + def build_akto_payload( + self, + inputs: GenericGuardrailAPIInputs, + request_data: dict, + *, + status_code: int = 200, + include_response: bool = False, + ) -> Dict[str, Any]: + """Build the flat MIRRORING payload sent to Akto's HTTP proxy endpoint. + + All body fields use double-encoding: json.dumps({"body": json.dumps(actual_body)}) + to match the canonical CLI hook format. + """ + request_path = self.extract_request_path(request_data) + request_headers = self.build_request_headers(request_data) + request_body = self.build_request_body(inputs, request_data) + tag = self.build_tag_metadata(request_data) + + response_payload = json.dumps({}) # Empty body wrapper when no response yet + response_headers: Dict[str, str] = {} + if include_response: + response_body = self.build_response_body(inputs, request_data) + response_payload = json.dumps({"body": json.dumps(response_body)}) # Double-encoded + response_headers = {"content-type": "application/json"} + + # Extract client IP from proxy headers + ip = "" + proxy_req = request_data.get("proxy_server_request", {}) + proxy_headers = proxy_req.get("headers", {}) if isinstance(proxy_req, dict) else {} + if isinstance(proxy_headers, dict): + ip = proxy_headers.get("x-forwarded-for") or proxy_headers.get("x-real-ip") or "" + if "," in ip: + ip = ip.split(",")[0].strip() + + return { + "path": request_path, + "requestHeaders": json.dumps(request_headers), + "responseHeaders": json.dumps(response_headers), + "method": "POST", + "requestPayload": json.dumps({"body": json.dumps(request_body)}), # Double-encoded + "responsePayload": response_payload, + "ip": ip, + "destIp": "127.0.0.1", + "time": str(int(datetime.now().timestamp() * 1000)), + "statusCode": str(status_code), + "type": "HTTP/1.1", + "status": str(status_code), + "akto_account_id": self.akto_account_id, + "akto_vxlan_id": self.akto_vxlan_id, + "is_pending": "false", + "source": "MIRRORING", + "direction": None, + "process_id": None, + "socket_id": None, + "daemonset_id": None, + "enabled_graph": None, + "tag": json.dumps(tag), + "metadata": json.dumps(tag), + "contextSource": "AGENTIC", + } + + async def send_request( + self, + *, + guardrails: bool, + ingest_data: bool, + payload: dict, + ) -> httpx.Response: + """Send an HTTP POST to the Akto API endpoint.""" + endpoint = f"{self.akto_base_url}{HTTP_PROXY_PATH}" + params = self.build_query_params(guardrails=guardrails, ingest_data=ingest_data) + headers = self.prepare_headers() + return await self.async_handler.post( + url=endpoint, + data=json.dumps(payload), + params=params, + headers=headers, + timeout=self.guardrail_timeout, + ) + + @staticmethod + def handle_guardrail_response(response: httpx.Response) -> Tuple[bool, str]: + """Parse the Akto guardrail response. Returns (allowed, reason).""" + if response.status_code != 200: + verbose_proxy_logger.error("Akto returned HTTP %d", response.status_code) + raise httpx.HTTPStatusError( + f"Akto returned unexpected status {response.status_code}", + request=response.request, + response=response, + ) + try: + result = response.json() + except (json.JSONDecodeError, ValueError) as e: + response_text = getattr(response, "text", "") + verbose_proxy_logger.error( + "Akto returned non-JSON body for status 200: %r", + response_text[:200], + ) + raise httpx.RequestError( + "Akto returned non-JSON body", + request=response.request, + ) from e + if not isinstance(result, dict): + return True, "" + data = result.get("data") or {} + if not isinstance(data, dict): + return True, "" + guardrails_result = data.get("guardrailsResult") or {} + if not isinstance(guardrails_result, dict): + return True, "" + return ( + bool(guardrails_result.get("Allowed", True)), + str(guardrails_result.get("Reason", "")), + ) + + def handle_unreachable( + self, + inputs: GenericGuardrailAPIInputs, + error: Exception, + ) -> GenericGuardrailAPIInputs: + """Handle Akto being unreachable based on fail_open/fail_closed config.""" + if self.unreachable_fallback == "fail_open": + verbose_proxy_logger.critical( + "Akto unreachable (fail-open): %s", + str(error), + exc_info=error, + ) + return inputs + + verbose_proxy_logger.error("Akto unreachable (fail-closed): %s", str(error)) + raise HTTPException( + status_code=503, + detail="Akto guardrail service unreachable", + ) + + async def fire_and_forget_request( + self, + *, + guardrails: bool, + ingest_data: bool, + payload: dict, + ) -> None: + """Send a request without awaiting it in the caller. Errors are logged, not raised.""" + try: + response = await self.send_request( + guardrails=guardrails, + ingest_data=ingest_data, + payload=payload, + ) + if response.status_code != 200: + verbose_proxy_logger.error( + "Akto fire-and-forget returned HTTP %d", + response.status_code, + ) + except Exception as e: + verbose_proxy_logger.error("Akto fire-and-forget error: %s", str(e)) + + @log_guardrail_information + async def apply_guardrail( + self, + inputs: GenericGuardrailAPIInputs, + request_data: dict, + input_type: Literal["request", "response"], + logging_obj=None, + ) -> GenericGuardrailAPIInputs: + """Main entry point called by LiteLLM's guardrail framework. + + Pre_call (input_type="request"): + - Awaits guardrail check. If blocked, fires off ingest with 403 marker and raises. + Post_call (input_type="response"): + - Fire-and-forget combined guardrail + ingest call. + """ + # Skip if this hook doesn't handle the current input_type + expected = self.HOOK_TO_INPUT.get(str(self.event_hook)) + if expected and expected != input_type: + return inputs + + if input_type == "request": + # Pre_call: awaited guardrail check (no ingestion) + payload = self.build_akto_payload(inputs, request_data, include_response=False) + try: + response = await self.send_request( + guardrails=True, + ingest_data=False, + payload=payload, + ) + allowed, reason = self.handle_guardrail_response(response) + except HTTPException: + raise + except (httpx.RequestError, httpx.HTTPStatusError) as e: + return self.handle_unreachable( + inputs=inputs, + error=e, + ) + + if not allowed: + # Build a blocked marker payload with 403 status and reason + blocked_payload = self.build_akto_payload( + inputs, + request_data, + include_response=False, + status_code=403, + ) + blocked_payload["responsePayload"] = json.dumps( + { + "body": json.dumps({"x-blocked-by": "Akto Proxy", "reason": reason}), + } + ) + blocked_payload["responseHeaders"] = json.dumps( + {"content-type": "application/json"}, + ) + # Fire-and-forget ingest of the blocked request, then raise 403 + task = asyncio.create_task( + self.fire_and_forget_request( + guardrails=False, + ingest_data=True, + payload=blocked_payload, + ) + ) + self.background_tasks.add(task) + task.add_done_callback(self.background_tasks.discard) + raise HTTPException( + status_code=403, + detail=reason or "Blocked by Akto Guardrails", + ) + + elif input_type == "response": + # Post_call: fire-and-forget combined guardrail + ingest + payload = self.build_akto_payload(inputs, request_data, include_response=True) + task = asyncio.create_task( + self.fire_and_forget_request( + guardrails=True, + ingest_data=True, + payload=payload, + ) + ) + self.background_tasks.add(task) + task.add_done_callback(self.background_tasks.discard) + + return inputs diff --git a/litellm/proxy/guardrails/guardrail_hooks/dynamoai/__init__.py b/litellm/proxy/guardrails/guardrail_hooks/dynamoai/__init__.py index 79f1992da4..f9ebf46a27 100644 --- a/litellm/proxy/guardrails/guardrail_hooks/dynamoai/__init__.py +++ b/litellm/proxy/guardrails/guardrail_hooks/dynamoai/__init__.py @@ -1,3 +1,33 @@ +from typing import TYPE_CHECKING + +from litellm.types.guardrails import SupportedGuardrailIntegrations + from .dynamoai import DynamoAIGuardrails -__all__ = ["DynamoAIGuardrails"] +if TYPE_CHECKING: + from litellm.types.guardrails import Guardrail, LitellmParams + + +def initialize_guardrail(litellm_params: "LitellmParams", guardrail: "Guardrail"): + import litellm + + _dynamoai_callback = DynamoAIGuardrails( + api_base=litellm_params.api_base, + api_key=litellm_params.api_key, + guardrail_name=guardrail.get("guardrail_name", ""), + event_hook=litellm_params.mode, + default_on=litellm_params.default_on, + ) + litellm.logging_callback_manager.add_litellm_callback(_dynamoai_callback) + + return _dynamoai_callback + + +guardrail_initializer_registry = { + SupportedGuardrailIntegrations.DYNAMOAI.value: initialize_guardrail, +} + + +guardrail_class_registry = { + SupportedGuardrailIntegrations.DYNAMOAI.value: DynamoAIGuardrails, +} diff --git a/litellm/types/guardrails.py b/litellm/types/guardrails.py index d5abf5c8fb..11b1d0d40c 100644 --- a/litellm/types/guardrails.py +++ b/litellm/types/guardrails.py @@ -17,6 +17,9 @@ from litellm.types.proxy.guardrails.guardrail_hooks.grayswan import ( from litellm.types.proxy.guardrails.guardrail_hooks.ibm import ( IBMGuardrailsBaseConfigModel, ) +from litellm.types.proxy.guardrails.guardrail_hooks.akto import ( + AktoConfigModel, +) from litellm.types.proxy.guardrails.guardrail_hooks.litellm_content_filter import ( ContentFilterCategoryConfig, ) @@ -33,7 +36,7 @@ Pydantic object defining how to set guardrails on litellm proxy guardrails: - guardrail_name: "bedrock-pre-guard" litellm_params: - guardrail: bedrock # supported values: "aporia", "bedrock", "lakera", "zscaler_ai_guard" + guardrail: bedrock # supported values: "akto", "aporia", "bedrock", "lakera", "zscaler_ai_guard" mode: "during_call" guardrailIdentifier: ff6ujrregl1q guardrailVersion: "DRAFT" @@ -44,6 +47,7 @@ guardrails: class SupportedGuardrailIntegrations(Enum): APORIA = "aporia" BEDROCK = "bedrock" + DYNAMOAI = "dynamoai" GUARDRAILS_AI = "guardrails_ai" LAKERA = "lakera" LAKERA_V2 = "lakera_v2" @@ -78,6 +82,7 @@ class SupportedGuardrailIntegrations(Enum): SEMANTIC_GUARD = "semantic_guard" MCP_END_USER_PERMISSION = "mcp_end_user_permission" BLOCK_CODE_EXECUTION = "block_code_execution" + AKTO = "akto" class Role(Enum): @@ -735,6 +740,7 @@ class LitellmParams( NomaGuardrailConfigModel, ToolPermissionGuardrailConfigModel, ZscalerAIGuardConfigModel, + AktoConfigModel, JavelinGuardrailConfigModel, BaseLitellmParams, EnkryptAIGuardrailConfigs, diff --git a/litellm/types/proxy/guardrails/guardrail_hooks/akto.py b/litellm/types/proxy/guardrails/guardrail_hooks/akto.py new file mode 100644 index 0000000000..180c89e811 --- /dev/null +++ b/litellm/types/proxy/guardrails/guardrail_hooks/akto.py @@ -0,0 +1,55 @@ +from typing import Optional, Literal + +from pydantic import Field + +from .base import GuardrailConfigModel + + +class AktoConfigModel(GuardrailConfigModel): + """ + Config for the Akto guardrail. + + Use two separate config entries to control behaviour: + akto-validate (mode: pre_call) -> check guardrails, block if flagged + akto-ingest (mode: post_call) -> ingest request+response data + """ + + akto_base_url: Optional[str] = Field( + default=None, + description="Akto Guardrail API Base URL. Env: AKTO_GUARDRAIL_API_BASE.", + json_schema_extra={ + "examples": [ + "http://localhost:9090", + "https://akto-ingestion.example.com", + ] + }, + ) + + akto_api_key: Optional[str] = Field( + default=None, + description="API key for Akto. Env: AKTO_API_KEY.", + ) + + akto_account_id: Optional[str] = Field( + default=None, + description="Akto account ID for multi-tenant deployments. Env: AKTO_ACCOUNT_ID. Default: '1000000'.", + ) + + akto_vxlan_id: Optional[str] = Field( + default=None, + description="Akto VXLAN ID. Env: AKTO_VXLAN_ID. Default: '0'.", + ) + + unreachable_fallback: Literal["fail_closed", "fail_open"] = Field( + default="fail_closed", + description="What to do when Akto is unreachable. 'fail_open' = allow, 'fail_closed' = block.", + ) + + guardrail_timeout: Optional[int] = Field( + default=None, + description="HTTP timeout in seconds. Default: 5.", + ) + + @staticmethod + def ui_friendly_name() -> str: + return "Akto" diff --git a/tests/guardrails_tests/test_akto_guardrails.py b/tests/guardrails_tests/test_akto_guardrails.py new file mode 100644 index 0000000000..3c70104a21 --- /dev/null +++ b/tests/guardrails_tests/test_akto_guardrails.py @@ -0,0 +1,550 @@ +import asyncio +import json +import os +from unittest.mock import AsyncMock, MagicMock, patch + +import httpx +import pytest + +from starlette.exceptions import HTTPException +from litellm.types.utils import GenericGuardrailAPIInputs +from litellm.proxy.guardrails.guardrail_registry import guardrail_initializer_registry, guardrail_class_registry +from litellm.proxy.guardrails.guardrail_hooks.akto.akto import AktoGuardrail + + +# --------------------------------------------------------------------------- +# Registry tests +# --------------------------------------------------------------------------- + + +def test_akto_in_guardrail_initializer_registry(): + assert "akto" in guardrail_initializer_registry + + +def test_akto_in_guardrail_class_registry(): + assert "akto" in guardrail_class_registry + assert guardrail_class_registry["akto"] is AktoGuardrail + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture +def akto_validate(): + """AktoGuardrail configured for pre_call (akto-validate).""" + return AktoGuardrail( + akto_base_url="http://localhost:9090", + akto_api_key="test-token", + unreachable_fallback="fail_closed", + guardrail_name="test-akto-validate", + event_hook="pre_call", + ) + + +@pytest.fixture +def akto_ingest(): + """AktoGuardrail configured for post_call (akto-ingest).""" + return AktoGuardrail( + akto_base_url="http://localhost:9090", + akto_api_key="test-token", + unreachable_fallback="fail_open", + guardrail_name="test-akto-ingest", + event_hook="post_call", + ) + + +@pytest.fixture +def sample_inputs() -> GenericGuardrailAPIInputs: + return GenericGuardrailAPIInputs( + texts=["Hello, how are you?"], + model="gpt-4", + ) + + +@pytest.fixture +def sample_request_data() -> dict: + return { + "metadata": { + "user_api_key_request_route": "/v1/chat/completions", + "user_api_key": "sk-test-123", + "user_api_key_user_id": "user-1", + "user_api_key_team_id": "team-1", + }, + "proxy_server_request": { + "headers": { + "x-forwarded-for": "10.0.0.1", + } + }, + } + + +def _mock_allowed_response(): + mock = MagicMock(spec=httpx.Response) + mock.status_code = 200 + mock.json.return_value = {"data": {"guardrailsResult": {"Allowed": True, "Reason": ""}}} + return mock + + +def _mock_blocked_response(reason="Prompt injection detected"): + mock = MagicMock(spec=httpx.Response) + mock.status_code = 200 + mock.json.return_value = {"data": {"guardrailsResult": {"Allowed": False, "Reason": reason}}} + return mock + + +# --------------------------------------------------------------------------- +# Initialization tests +# --------------------------------------------------------------------------- + + +def test_init_requires_akto_base_url(): + with patch.dict(os.environ, {}, clear=True): + with pytest.raises(ValueError, match="akto_base_url is required"): + AktoGuardrail( + akto_base_url="", + akto_api_key="test-token", + guardrail_name="test", + event_hook="pre_call", + ) + + +def test_init_requires_api_key(): + with patch.dict(os.environ, {}, clear=True): + with pytest.raises(ValueError, match="akto_api_key is required"): + AktoGuardrail( + akto_base_url="http://localhost:9090", + akto_api_key="", + guardrail_name="test", + event_hook="pre_call", + ) + + +def test_init_from_env(): + with patch.dict( + os.environ, + { + "AKTO_GUARDRAIL_API_BASE": "http://env-host:9090", + "AKTO_API_KEY": "env-token", + "AKTO_ACCOUNT_ID": "2000000", + "AKTO_VXLAN_ID": "42", + }, + ): + g = AktoGuardrail(guardrail_name="env-test", event_hook="post_call") + assert g.akto_base_url == "http://env-host:9090" + assert g.akto_api_key == "env-token" + assert g.guardrail_timeout == 5 + assert g.akto_account_id == "2000000" + assert g.akto_vxlan_id == "42" + + +def test_init_defaults(): + g = AktoGuardrail( + akto_base_url="http://localhost:9090", + akto_api_key="test-token", + guardrail_name="default-test", + event_hook="pre_call", + ) + assert g.unreachable_fallback == "fail_closed" + assert g.guardrail_timeout == 5 + assert g.akto_account_id == "1000000" + assert g.akto_vxlan_id == "0" + + +def test_background_tasks_per_instance(): + a = AktoGuardrail( + akto_base_url="http://localhost:9090", + akto_api_key="test-token", + guardrail_name="instance-a", + event_hook="pre_call", + ) + b = AktoGuardrail( + akto_base_url="http://localhost:9090", + akto_api_key="test-token", + guardrail_name="instance-b", + event_hook="post_call", + ) + assert a.background_tasks is not b.background_tasks + + +# --------------------------------------------------------------------------- +# Payload format tests +# --------------------------------------------------------------------------- + + +def test_build_akto_payload_format(akto_validate, sample_inputs, sample_request_data): + payload = akto_validate.build_akto_payload(sample_inputs, sample_request_data, include_response=False) + + assert payload["path"] == "/v1/chat/completions" + assert payload["method"] == "POST" + assert payload["type"] == "HTTP/1.1" + assert payload["akto_account_id"] == "1000000" + assert payload["akto_vxlan_id"] == "0" + assert payload["is_pending"] == "false" + assert payload["source"] == "MIRRORING" + assert payload["contextSource"] == "AGENTIC" + assert payload["ip"] == "10.0.0.1" + + req_headers = json.loads(payload["requestHeaders"]) + assert "content-type" in req_headers + + req_wrapper = json.loads(payload["requestPayload"]) + req_body = json.loads(req_wrapper["body"]) + assert req_body["model"] == "gpt-4" + assert req_body["messages"][0]["content"] == "Hello, how are you?" + + tag = json.loads(payload["tag"]) + assert tag["gen-ai"] == "Gen AI" + + assert payload["responsePayload"] == json.dumps({}) + assert payload["time"].isdigit() + assert len(payload["time"]) >= 13 + + +def test_build_akto_payload_with_response(akto_validate, sample_inputs, sample_request_data): + payload = akto_validate.build_akto_payload(sample_inputs, sample_request_data, include_response=True) + resp_wrapper = json.loads(payload["responsePayload"]) + resp_body = json.loads(resp_wrapper["body"]) + assert "choices" in resp_body + + +def test_build_akto_payload_custom_account_ids(sample_inputs, sample_request_data): + g = AktoGuardrail( + akto_base_url="http://localhost:9090", + akto_api_key="test-token", + akto_account_id="9999", + akto_vxlan_id="7", + guardrail_name="custom-ids-test", + event_hook="pre_call", + ) + payload = g.build_akto_payload(sample_inputs, sample_request_data, include_response=False) + assert payload["akto_account_id"] == "9999" + assert payload["akto_vxlan_id"] == "7" + + +def test_build_query_params(): + params = AktoGuardrail.build_query_params(guardrails=True, ingest_data=False) + assert params == {"akto_connector": "litellm", "guardrails": "true"} + + params = AktoGuardrail.build_query_params(guardrails=False, ingest_data=True) + assert params == {"akto_connector": "litellm", "ingest_data": "true"} + + params = AktoGuardrail.build_query_params(guardrails=True, ingest_data=True) + assert params == { + "akto_connector": "litellm", + "guardrails": "true", + "ingest_data": "true", + } + + +# --------------------------------------------------------------------------- +# Guardrail response handling +# --------------------------------------------------------------------------- + + +def test_handle_guardrail_response_allowed(): + mock_resp = MagicMock(spec=httpx.Response) + mock_resp.status_code = 200 + mock_resp.json.return_value = {"data": {"guardrailsResult": {"Allowed": True, "Reason": ""}}} + allowed, reason = AktoGuardrail.handle_guardrail_response(mock_resp) + assert allowed is True + assert reason == "" + + +def test_handle_guardrail_response_blocked(): + mock_resp = MagicMock(spec=httpx.Response) + mock_resp.status_code = 200 + mock_resp.json.return_value = {"data": {"guardrailsResult": {"Allowed": False, "Reason": "PII detected"}}} + allowed, reason = AktoGuardrail.handle_guardrail_response(mock_resp) + assert allowed is False + assert reason == "PII detected" + + +def test_handle_guardrail_response_missing_result(): + mock_resp = MagicMock(spec=httpx.Response) + mock_resp.status_code = 200 + mock_resp.json.return_value = {} + allowed, _ = AktoGuardrail.handle_guardrail_response(mock_resp) + assert allowed is True + + +def test_handle_guardrail_response_data_none(): + mock_resp = MagicMock(spec=httpx.Response) + mock_resp.status_code = 200 + mock_resp.json.return_value = {"data": None} + allowed, reason = AktoGuardrail.handle_guardrail_response(mock_resp) + assert allowed is True + assert reason == "" + + +def test_handle_guardrail_response_guardrails_result_not_dict(): + mock_resp = MagicMock(spec=httpx.Response) + mock_resp.status_code = 200 + mock_resp.json.return_value = {"data": {"guardrailsResult": "invalid"}} + allowed, reason = AktoGuardrail.handle_guardrail_response(mock_resp) + assert allowed is True + assert reason == "" + + +def test_handle_guardrail_response_non_dict(): + mock_resp = MagicMock(spec=httpx.Response) + mock_resp.status_code = 200 + mock_resp.json.return_value = "invalid" + allowed, _ = AktoGuardrail.handle_guardrail_response(mock_resp) + assert allowed is True + + +def test_handle_guardrail_response_error_status(): + mock_resp = MagicMock(spec=httpx.Response) + mock_resp.status_code = 500 + mock_resp.request = MagicMock() + with pytest.raises(httpx.HTTPStatusError): + AktoGuardrail.handle_guardrail_response(mock_resp) + + +def test_handle_guardrail_response_non_json_body(): + mock_resp = MagicMock(spec=httpx.Response) + mock_resp.status_code = 200 + mock_resp.request = MagicMock() + mock_resp.text = "not json" + mock_resp.json.side_effect = json.JSONDecodeError("Expecting value", "", 0) + + with pytest.raises(httpx.RequestError): + AktoGuardrail.handle_guardrail_response(mock_resp) + + +# --------------------------------------------------------------------------- +# Pre-call (akto-validate) — allowed +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_pre_call_allowed(akto_validate, sample_inputs, sample_request_data): + akto_validate.async_handler.post = AsyncMock(return_value=_mock_allowed_response()) + + result = await akto_validate.apply_guardrail( + inputs=sample_inputs, + request_data=sample_request_data, + input_type="request", + ) + + assert result == sample_inputs + akto_validate.async_handler.post.assert_called_once() + call_params = akto_validate.async_handler.post.call_args.kwargs["params"] + assert call_params.get("guardrails") == "true" + assert "ingest_data" not in call_params + + +# --------------------------------------------------------------------------- +# Pre-call (akto-validate) — blocked +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_pre_call_blocked(akto_validate, sample_inputs, sample_request_data): + akto_validate.async_handler.post = AsyncMock( + side_effect=[ + _mock_blocked_response("PII detected"), + _mock_allowed_response(), + ] + ) + + with pytest.raises(HTTPException) as exc_info: + await akto_validate.apply_guardrail( + inputs=sample_inputs, + request_data=sample_request_data, + input_type="request", + ) + + await asyncio.sleep(0) + await asyncio.sleep(0) + + assert exc_info.value.status_code == 403 + + assert akto_validate.async_handler.post.call_count == 2 + + first_call_params = akto_validate.async_handler.post.call_args_list[0].kwargs["params"] + assert first_call_params.get("guardrails") == "true" + + second_call_params = akto_validate.async_handler.post.call_args_list[1].kwargs["params"] + assert second_call_params.get("ingest_data") == "true" + assert "guardrails" not in second_call_params + second_payload = json.loads(akto_validate.async_handler.post.call_args_list[1].kwargs["data"]) + assert second_payload["statusCode"] == "403" + resp_body = json.loads(second_payload["responsePayload"]) + inner = json.loads(resp_body["body"]) + assert inner["x-blocked-by"] == "Akto Proxy" + assert inner["reason"] == "PII detected" + + +# --------------------------------------------------------------------------- +# Pre-call (akto-validate) — response input is no-op +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_validate_response_noop(akto_validate, sample_inputs, sample_request_data): + akto_validate.async_handler.post = AsyncMock() + + result = await akto_validate.apply_guardrail( + inputs=sample_inputs, + request_data=sample_request_data, + input_type="response", + ) + + assert result == sample_inputs + akto_validate.async_handler.post.assert_not_called() + + +# --------------------------------------------------------------------------- +# Post-call (akto-ingest) — combined guardrail + ingest +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_post_call_combined(akto_ingest, sample_inputs, sample_request_data): + akto_ingest.async_handler.post = AsyncMock(return_value=_mock_allowed_response()) + + result = await akto_ingest.apply_guardrail( + inputs=sample_inputs, + request_data=sample_request_data, + input_type="response", + ) + + await asyncio.sleep(0) + await asyncio.sleep(0) + + assert result == sample_inputs + akto_ingest.async_handler.post.assert_called_once() + call_params = akto_ingest.async_handler.post.call_args.kwargs["params"] + assert call_params.get("guardrails") == "true" + assert call_params.get("ingest_data") == "true" + + +# --------------------------------------------------------------------------- +# Post-call (akto-ingest) — request input is no-op +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_ingest_request_noop(akto_ingest, sample_inputs, sample_request_data): + akto_ingest.async_handler.post = AsyncMock() + + result = await akto_ingest.apply_guardrail( + inputs=sample_inputs, + request_data=sample_request_data, + input_type="request", + ) + + assert result == sample_inputs + akto_ingest.async_handler.post.assert_not_called() + + +# --------------------------------------------------------------------------- +# Fail-open / fail-closed +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_fail_open_on_unreachable(): + g = AktoGuardrail( + akto_base_url="http://localhost:9090", + akto_api_key="test-token", + unreachable_fallback="fail_open", + guardrail_name="fail-open-test", + event_hook="pre_call", + ) + g.async_handler.post = AsyncMock(side_effect=httpx.ConnectError("Connection refused")) + + inputs = GenericGuardrailAPIInputs(texts=["test"], model="gpt-4") + result = await g.apply_guardrail(inputs=inputs, request_data={}, input_type="request") + + assert result.get("texts") == ["test"] + + +@pytest.mark.asyncio +async def test_fail_closed_on_unreachable(): + g = AktoGuardrail( + akto_base_url="http://localhost:9090", + akto_api_key="test-token", + unreachable_fallback="fail_closed", + guardrail_name="fail-closed-test", + event_hook="pre_call", + ) + g.async_handler.post = AsyncMock(side_effect=httpx.ConnectError("Connection refused")) + + inputs = GenericGuardrailAPIInputs(texts=["test"], model="gpt-4") + with pytest.raises(HTTPException) as exc_info: + await g.apply_guardrail(inputs=inputs, request_data={}, input_type="request") + assert exc_info.value.status_code == 503 + + +def test_fail_closed_generic_message(): + g = AktoGuardrail( + akto_base_url="http://localhost:9090", + akto_api_key="test-token", + unreachable_fallback="fail_closed", + guardrail_name="msg-test", + event_hook="pre_call", + ) + with pytest.raises(HTTPException) as exc_info: + g.handle_unreachable( + inputs=GenericGuardrailAPIInputs(texts=["test"], model="gpt-4"), + error=Exception("http://internal-host:9090/secret-path"), + ) + assert "internal-host" not in exc_info.value.detail + assert exc_info.value.detail == "Akto guardrail service unreachable" + + +# --------------------------------------------------------------------------- +# Helper method tests +# --------------------------------------------------------------------------- + + +def test_extract_request_path_from_metadata(): + path = AktoGuardrail.extract_request_path({"metadata": {"user_api_key_request_route": "/v1/embeddings"}}) + assert path == "/v1/embeddings" + + +def test_extract_request_path_fallback(): + path = AktoGuardrail.extract_request_path({}) + assert path == "/v1/chat/completions" + + +def test_extract_request_path_non_dict_metadata(): + path = AktoGuardrail.extract_request_path({"metadata": "invalid"}) + assert path == "/v1/chat/completions" + + +def test_resolve_metadata_value(): + assert ( + AktoGuardrail.resolve_metadata_value({"metadata": {"user_api_key_user_id": "u1"}}, "user_api_key_user_id") + == "u1" + ) + assert ( + AktoGuardrail.resolve_metadata_value( + {"litellm_metadata": {"user_api_key_team_id": "t1"}}, + "user_api_key_team_id", + ) + == "t1" + ) + assert AktoGuardrail.resolve_metadata_value({}, "some_key") is None + assert AktoGuardrail.resolve_metadata_value(None, "some_key") is None + + +def test_resolve_metadata_value_non_dict_containers(): + assert ( + AktoGuardrail.resolve_metadata_value( + {"metadata": "invalid", "litellm_metadata": ["bad"]}, + "some_key", + ) + is None + ) + + +def test_build_tag_metadata(akto_validate, sample_request_data): + tag = akto_validate.build_tag_metadata(sample_request_data) + assert tag["gen-ai"] == "Gen AI" + assert tag["user_id"] == "user-1" + assert tag["team_id"] == "team-1" diff --git a/tests/test_litellm/proxy/guardrails/guardrail_hooks/test_dynamoai.py b/tests/test_litellm/proxy/guardrails/guardrail_hooks/test_dynamoai.py new file mode 100644 index 0000000000..7bc4e951a5 --- /dev/null +++ b/tests/test_litellm/proxy/guardrails/guardrail_hooks/test_dynamoai.py @@ -0,0 +1,81 @@ +""" +Tests for DynamoAI guardrail registration and initialization. +""" + +import os +from unittest.mock import patch + +import pytest + + +class TestDynamoAIGuardrailRegistration: + """Tests for DynamoAI guardrail registration in the guardrail system.""" + + def test_supported_guardrail_enum_entry(self): + """Test that DYNAMOAI is in SupportedGuardrailIntegrations enum.""" + from litellm.types.guardrails import SupportedGuardrailIntegrations + + assert hasattr(SupportedGuardrailIntegrations, "DYNAMOAI") + assert SupportedGuardrailIntegrations.DYNAMOAI.value == "dynamoai" + + def test_initialize_guardrail_function_exists(self): + """Test that initialize_guardrail function is properly exported.""" + from litellm.proxy.guardrails.guardrail_hooks.dynamoai import ( + guardrail_initializer_registry, + initialize_guardrail, + ) + + assert initialize_guardrail is not None + assert "dynamoai" in guardrail_initializer_registry + + def test_guardrail_class_registry_exists(self): + """Test that guardrail_class_registry is properly exported.""" + from litellm.proxy.guardrails.guardrail_hooks.dynamoai import ( + guardrail_class_registry, + ) + from litellm.proxy.guardrails.guardrail_hooks.dynamoai.dynamoai import ( + DynamoAIGuardrails, + ) + + assert "dynamoai" in guardrail_class_registry + assert guardrail_class_registry["dynamoai"] == DynamoAIGuardrails + + def test_initialize_guardrail_creates_instance(self): + """Test that initialize_guardrail creates a DynamoAIGuardrails instance.""" + from litellm.proxy.guardrails.guardrail_hooks.dynamoai import ( + initialize_guardrail, + ) + from litellm.proxy.guardrails.guardrail_hooks.dynamoai.dynamoai import ( + DynamoAIGuardrails, + ) + from litellm.types.guardrails import LitellmParams + + litellm_params = LitellmParams( + guardrail="dynamoai", + mode="pre_call", + api_key="test-key", + api_base="https://test.dynamo.ai", + ) + + guardrail = { + "guardrail_name": "test-dynamoai-guard", + } + + with patch( + "litellm.logging_callback_manager.add_litellm_callback" + ) as mock_add: + result = initialize_guardrail(litellm_params, guardrail) + + assert isinstance(result, DynamoAIGuardrails) + assert result.api_key == "test-key" + assert result.api_base == "https://test.dynamo.ai" + assert result.guardrail_name == "test-dynamoai-guard" + mock_add.assert_called_once_with(result) + + def test_dynamoai_in_global_registry(self): + """Test that dynamoai is discoverable in the global guardrail registry.""" + from litellm.proxy.guardrails.guardrail_registry import ( + guardrail_initializer_registry, + ) + + assert "dynamoai" in guardrail_initializer_registry diff --git a/ui/litellm-dashboard/public/assets/logos/akto.svg b/ui/litellm-dashboard/public/assets/logos/akto.svg new file mode 100644 index 0000000000..cdea32535f --- /dev/null +++ b/ui/litellm-dashboard/public/assets/logos/akto.svg @@ -0,0 +1,10 @@ + + + + + + + + + + diff --git a/ui/litellm-dashboard/src/components/guardrails/guardrail_garden_configs.ts b/ui/litellm-dashboard/src/components/guardrails/guardrail_garden_configs.ts index 7a1b5314d3..e42ecaef57 100644 --- a/ui/litellm-dashboard/src/components/guardrails/guardrail_garden_configs.ts +++ b/ui/litellm-dashboard/src/components/guardrails/guardrail_garden_configs.ts @@ -264,4 +264,10 @@ export const GUARDRAIL_PRESETS: Record = { mode: "pre_call", defaultOn: false, }, + akto: { + provider: "Akto", + guardrailNameSuggestion: "Akto Guardrail", + mode: "pre_call", + defaultOn: false, + }, }; diff --git a/ui/litellm-dashboard/src/components/guardrails/guardrail_garden_data.ts b/ui/litellm-dashboard/src/components/guardrails/guardrail_garden_data.ts index 53ccb32c18..b06400ce50 100644 --- a/ui/litellm-dashboard/src/components/guardrails/guardrail_garden_data.ts +++ b/ui/litellm-dashboard/src/components/guardrails/guardrail_garden_data.ts @@ -373,6 +373,14 @@ export const PARTNER_GUARDRAIL_CARDS: GuardrailCardInfo[] = [ logo: `${ASSET_PREFIX}pillar.jpeg`, tags: ["Monitoring", "Safety"], }, + { + id: "akto", + name: "Akto Guardrail", + description: "AI security platform from Akto.io with automatic monitoring and guardrails for AI/ML applications.", + category: "partner", + logo: `${ASSET_PREFIX}akto.svg`, + tags: ["Security", "Safety", "Monitoring"], + }, ]; export const ALL_CARDS = [...LITELLM_CONTENT_FILTER_CARDS, ...PARTNER_GUARDRAIL_CARDS]; diff --git a/ui/litellm-dashboard/src/components/guardrails/guardrail_info_helpers.tsx b/ui/litellm-dashboard/src/components/guardrails/guardrail_info_helpers.tsx index d957be4306..c78835dae0 100644 --- a/ui/litellm-dashboard/src/components/guardrails/guardrail_info_helpers.tsx +++ b/ui/litellm-dashboard/src/components/guardrails/guardrail_info_helpers.tsx @@ -125,6 +125,7 @@ export const guardrailLogoMap: Record = { EnkryptAI: `${asset_logos_folder}enkrypt_ai.avif`, "Prompt Security": `${asset_logos_folder}prompt_security.png`, "LiteLLM Content Filter": `${asset_logos_folder}litellm_logo.jpg`, + "Akto": `${asset_logos_folder}akto.svg`, }; export const getGuardrailLogoAndName = (guardrailValue: string): { logo: string; displayName: string } => {