Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/pass_through_endpoints/passthrough_guardrails.py: 29%
102 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1"""
2Passthrough Guardrails Helper Module
4Handles guardrail execution for passthrough endpoints with:
5- Opt-in model (guardrails only run when explicitly configured)
6- Field-level targeting using JSONPath expressions
7- Automatic inheritance from org/team/key levels when enabled
8"""
10from typing import Any, Final
12from litellm._logging import verbose_proxy_logger
13from litellm.proxy._types import (
14 PassThroughGuardrailsConfig,
15 PassThroughGuardrailSettings,
16 UserAPIKeyAuth,
17)
18from litellm.proxy.pass_through_endpoints.jsonpath_extractor import JsonPathExtractor
20# Type for raw guardrails config input (before normalization)
21# Can be a list of names or a dict with settings
22PassThroughGuardrailsConfigInput = (
23 list[str] # Simple list: ["guard-1", "guard-2"]
24 | PassThroughGuardrailsConfig # Dict: {"guard-1": {"request_fields": [...]}}
25)
28class PassthroughGuardrailHandler:
29 """
30 Handles guardrail execution for passthrough endpoints.
32 Passthrough endpoints use an opt-in model for guardrails:
33 - Guardrails only run when explicitly configured on the endpoint
34 - Supports field-level targeting using JSONPath expressions
35 - Automatically inherits org/team/key level guardrails when enabled
37 Guardrails can be specified as:
38 - List format (simple): ["guardrail-1", "guardrail-2"]
39 - Dict format (with settings): {"guardrail-1": {"request_fields": ["query"]}}
40 """
42 @staticmethod
43 def normalize_config(
44 guardrails_config: PassThroughGuardrailsConfigInput | None,
45 ) -> PassThroughGuardrailsConfig | None:
46 """
47 Normalize guardrails config to dict format.
49 Accepts:
50 - List of guardrail names: ["g1", "g2"] -> {"g1": None, "g2": None}
51 - Dict with settings: {"g1": {"request_fields": [...]}}
52 - None: returns None
53 """
54 if guardrails_config is None:
55 return None
57 # Already a dict - return as-is
58 if isinstance(guardrails_config, dict): 58 ↛ 62line 58 didn't jump to line 62 because the condition on line 58 was always true
59 return guardrails_config
61 # List of guardrail names - convert to dict
62 if isinstance(guardrails_config, list):
63 return {name: None for name in guardrails_config}
65 verbose_proxy_logger.debug(
66 "Passthrough guardrails config is not a dict or list, got: %s",
67 type(guardrails_config),
68 )
69 return None
71 @staticmethod
72 def is_enabled(
73 guardrails_config: PassThroughGuardrailsConfigInput | None,
74 ) -> bool:
75 """
76 Check if guardrails are enabled for a passthrough endpoint.
78 Passthrough endpoints are opt-in only - guardrails only run when
79 the guardrails config is set with at least one guardrail.
80 """
81 normalized: Final = PassthroughGuardrailHandler.normalize_config(guardrails_config)
82 if normalized is None:
83 return False
84 return len(normalized) > 0
86 @staticmethod
87 def get_guardrail_names(
88 guardrails_config: PassThroughGuardrailsConfigInput | None,
89 ) -> list[str]:
90 """Get the list of guardrail names configured for a passthrough endpoint."""
91 normalized: Final = PassthroughGuardrailHandler.normalize_config(guardrails_config)
92 if normalized is None:
93 return []
94 return list(normalized.keys())
96 @staticmethod
97 def get_settings(
98 guardrails_config: PassThroughGuardrailsConfigInput | None,
99 guardrail_name: str,
100 ) -> PassThroughGuardrailSettings | None:
101 """Get settings for a specific guardrail from the passthrough config."""
102 normalized: Final = PassthroughGuardrailHandler.normalize_config(guardrails_config)
103 if normalized is None:
104 return None
106 settings: Final = normalized.get(guardrail_name)
107 if settings is None:
108 return None
110 if isinstance(settings, dict):
111 return PassThroughGuardrailSettings(**settings)
113 return settings
115 @staticmethod
116 def prepare_input(
117 request_data: dict,
118 guardrail_settings: PassThroughGuardrailSettings | None,
119 ) -> str:
120 """
121 Prepare input text for guardrail execution based on field targeting settings.
123 If request_fields is specified, extracts only those fields.
124 Otherwise, uses the entire request payload as text.
125 """
126 if guardrail_settings is None or guardrail_settings.request_fields is None:
127 from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
129 return safe_dumps(request_data)
131 return JsonPathExtractor.extract_fields(
132 data=request_data,
133 jsonpath_expressions=guardrail_settings.request_fields,
134 )
136 @staticmethod
137 def prepare_output(
138 response_data: dict,
139 guardrail_settings: PassThroughGuardrailSettings | None,
140 ) -> str:
141 """
142 Prepare output text for guardrail execution based on field targeting settings.
144 If response_fields is specified, extracts only those fields.
145 Otherwise, uses the entire response payload as text.
146 """
147 if guardrail_settings is None or guardrail_settings.response_fields is None:
148 from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
150 return safe_dumps(response_data)
152 return JsonPathExtractor.extract_fields(
153 data=response_data,
154 jsonpath_expressions=guardrail_settings.response_fields,
155 )
157 @staticmethod
158 async def execute(
159 request_data: dict,
160 user_api_key_dict: UserAPIKeyAuth,
161 guardrails_config: PassThroughGuardrailsConfig | None,
162 event_type: str = "pre_call",
163 ) -> dict:
164 """
165 Execute guardrails for a passthrough endpoint.
167 This is the main entry point for passthrough guardrail execution.
169 Args:
170 request_data: The request payload
171 user_api_key_dict: User API key authentication info
172 guardrails_config: Passthrough-specific guardrails configuration
173 event_type: "pre_call" for request, "post_call" for response
175 Returns:
176 The potentially modified request_data
178 Raises:
179 HTTPException if a guardrail blocks the request
180 """
181 if not PassthroughGuardrailHandler.is_enabled(guardrails_config):
182 verbose_proxy_logger.debug("Passthrough guardrails not enabled, skipping guardrail execution")
183 return request_data
185 guardrail_names: Final = PassthroughGuardrailHandler.get_guardrail_names(guardrails_config)
186 verbose_proxy_logger.debug("Executing passthrough guardrails: %s", guardrail_names)
188 # Add to request metadata so guardrails know which to run
189 from litellm.proxy.pass_through_endpoints.passthrough_context import (
190 set_passthrough_guardrails_config,
191 )
193 if "metadata" not in request_data:
194 request_data["metadata"] = {}
196 # Set guardrails in metadata using dict format for compatibility
197 request_data["metadata"]["guardrails"] = {name: True for name in guardrail_names}
199 # Store passthrough guardrails config in request-scoped context
200 set_passthrough_guardrails_config(guardrails_config)
202 return request_data
204 @staticmethod
205 def collect_guardrails(
206 user_api_key_dict: UserAPIKeyAuth,
207 passthrough_guardrails_config: PassThroughGuardrailsConfigInput | None,
208 ) -> dict[str, bool] | None:
209 """
210 Collect guardrails for a passthrough endpoint.
212 Passthrough endpoints are opt-in only for guardrails. Guardrails only run when
213 the guardrails config is set with at least one guardrail.
215 Accepts both list and dict formats:
216 - List: ["guardrail-1", "guardrail-2"]
217 - Dict: {"guardrail-1": {"request_fields": [...]}}
219 When enabled, this function collects:
220 - Passthrough-specific guardrails from the config
221 - Org/team/key level guardrails (automatic inheritance when passthrough is enabled)
223 Args:
224 user_api_key_dict: User API key authentication info
225 passthrough_guardrails_config: List or Dict of guardrail names/settings
227 Returns:
228 Dict of guardrail names to run (format: {guardrail_name: True}), or None
229 """
230 from litellm.proxy.litellm_pre_call_utils import (
231 _add_guardrails_from_key_or_team_metadata,
232 )
234 # Normalize config to dict format (handles both list and dict)
235 normalized_config: Final = PassthroughGuardrailHandler.normalize_config(passthrough_guardrails_config)
237 if normalized_config is None:
238 verbose_proxy_logger.debug("Passthrough guardrails not configured, skipping guardrail collection")
239 return None
241 if len(normalized_config) == 0: 241 ↛ 246line 241 didn't jump to line 246 because the condition on line 241 was always true
242 verbose_proxy_logger.debug("Passthrough guardrails config is empty, skipping")
243 return None
245 # Passthrough is enabled - collect guardrails
246 guardrails_to_run: Final[dict[str, bool]] = {}
248 # Add passthrough-specific guardrails
249 for guardrail_name in normalized_config:
250 guardrails_to_run[guardrail_name] = True
251 verbose_proxy_logger.debug("Added passthrough-specific guardrail: %s", guardrail_name)
253 # Add org/team/key level guardrails using shared helper
254 temp_data: Final[dict[str, Any]] = {"metadata": {}}
255 _add_guardrails_from_key_or_team_metadata(
256 key_metadata=user_api_key_dict.metadata,
257 team_metadata=user_api_key_dict.team_metadata,
258 data=temp_data,
259 metadata_variable_name="metadata",
260 )
262 # Merge inherited guardrails into guardrails_to_run
263 inherited_guardrails: Final = temp_data["metadata"].get("guardrails", [])
264 for guardrail_name in inherited_guardrails:
265 if guardrail_name not in guardrails_to_run:
266 guardrails_to_run[guardrail_name] = True
267 verbose_proxy_logger.debug("Added inherited guardrail (key/team level): %s", guardrail_name)
269 verbose_proxy_logger.debug(
270 "Collected guardrails for passthrough endpoint: %s",
271 list(guardrails_to_run.keys()),
272 )
274 return guardrails_to_run if guardrails_to_run else None
276 @staticmethod
277 def get_field_targeted_text(
278 data: dict,
279 guardrail_name: str,
280 is_request: bool = True,
281 ) -> str | None:
282 """
283 Get the text to check for a guardrail, respecting field targeting settings.
285 Called by guardrail hooks to get the appropriate text based on
286 passthrough field targeting configuration.
288 Args:
289 data: The request/response data dict
290 guardrail_name: Name of the guardrail being executed
291 is_request: True for request (pre_call), False for response (post_call)
293 Returns:
294 The text to check, or None to use default behavior
295 """
296 from litellm.proxy.pass_through_endpoints.passthrough_context import (
297 get_passthrough_guardrails_config,
298 )
300 passthrough_config: Final = get_passthrough_guardrails_config()
301 if passthrough_config is None:
302 return None
304 settings: Final = PassthroughGuardrailHandler.get_settings(passthrough_config, guardrail_name)
305 if settings is None:
306 return None
308 if is_request:
309 if settings.request_fields:
310 return JsonPathExtractor.extract_fields(data, settings.request_fields)
311 else:
312 if settings.response_fields:
313 return JsonPathExtractor.extract_fields(data, settings.response_fields)
315 return None