Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/policy_engine/policy_resolver.py: 79%
102 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1"""
2Policy Resolver - Resolves final guardrail list from policies.
4Handles:
5- Inheritance chain resolution (inherit with add/remove)
6- Applying add/remove guardrails
7- Evaluating model conditions
8- Combining guardrails from multiple matching policies
9"""
11from typing import Final
13from litellm._logging import verbose_proxy_logger
14from litellm.types.proxy.policy_engine import (
15 GuardrailPipeline,
16 Policy,
17 PolicyMatchContext,
18 ResolvedPolicy,
19)
22class PolicyResolver:
23 """
24 Resolves the final list of guardrails from policies.
26 Handles:
27 - Inheritance chains with add/remove operations
28 - Model-based conditions
29 """
31 @staticmethod
32 def resolve_inheritance_chain(
33 policy_name: str,
34 policies: dict[str, Policy],
35 visited: set[str] | None = None,
36 ) -> list[str]:
37 """
38 Get the inheritance chain for a policy (from root to policy).
40 Args:
41 policy_name: Name of the policy
42 policies: Dictionary of all policies
43 visited: Set of visited policies (for cycle detection)
45 Returns:
46 List of policy names from root ancestor to the given policy
47 """
48 if visited is None:
49 visited = set()
51 if policy_name in visited: 51 ↛ 52line 51 didn't jump to line 52 because the condition on line 51 was never true
52 verbose_proxy_logger.warning("Circular inheritance detected for policy '%s'", policy_name)
53 return []
55 policy: Final = policies.get(policy_name)
56 if policy is None:
57 return []
59 visited.add(policy_name)
61 if policy.inherit:
62 parent_chain: Final = PolicyResolver.resolve_inheritance_chain(
63 policy_name=policy.inherit, policies=policies, visited=visited
64 )
65 return parent_chain + [policy_name]
67 return [policy_name]
69 @staticmethod
70 def resolve_policy_guardrails(
71 policy_name: str,
72 policies: dict[str, Policy],
73 context: PolicyMatchContext | None = None,
74 ) -> ResolvedPolicy:
75 """
76 Resolve the final guardrails for a single policy, including inheritance.
78 This method:
79 1. Resolves the inheritance chain
80 2. Applies add/remove from each policy in the chain
81 3. Evaluates model conditions (if context provided)
83 Args:
84 policy_name: Name of the policy to resolve
85 policies: Dictionary of all policies
86 context: Optional request context for evaluating conditions
88 Returns:
89 ResolvedPolicy with final guardrails list
90 """
91 from litellm.proxy.policy_engine.condition_evaluator import ConditionEvaluator
93 inheritance_chain: Final = PolicyResolver.resolve_inheritance_chain(policy_name=policy_name, policies=policies)
95 # Start with empty set of guardrails
96 guardrails: Final[set[str]] = set()
98 # Apply each policy in the chain (from root to leaf)
99 for chain_policy_name in inheritance_chain:
100 policy = policies.get(chain_policy_name)
101 if policy is None: 101 ↛ 102line 101 didn't jump to line 102 because the condition on line 101 was never true
102 continue
104 # Check if policy condition matches (if context provided)
105 if context is not None and policy.condition is not None:
106 if not ConditionEvaluator.evaluate( 106 ↛ 110line 106 didn't jump to line 110 because the condition on line 106 was never true
107 condition=policy.condition,
108 context=context,
109 ):
110 verbose_proxy_logger.debug(
111 "Policy '%s' condition did not match, skipping guardrails", chain_policy_name
112 )
113 continue
115 # Add guardrails from guardrails.add
116 for guardrail in policy.guardrails.get_add():
117 guardrails.add(guardrail)
119 # Remove guardrails from guardrails.remove
120 for guardrail in policy.guardrails.get_remove():
121 guardrails.discard(guardrail)
123 return ResolvedPolicy(
124 policy_name=policy_name,
125 guardrails=list(guardrails),
126 inheritance_chain=inheritance_chain,
127 )
129 @staticmethod
130 def resolve_guardrails_for_context(
131 context: PolicyMatchContext,
132 policies: dict[str, Policy] | None = None,
133 policy_names: list[str] | None = None,
134 ) -> list[str]:
135 """
136 Resolve the final list of guardrails for a request context.
138 This:
139 1. Finds all policies that match the context via policy_attachments (or policy_names if provided)
140 2. Resolves each policy's guardrails (including inheritance)
141 3. Evaluates model conditions
142 4. Combines all guardrails (union)
144 Args:
145 context: The request context
146 policies: Dictionary of all policies (if None, uses global registry)
147 policy_names: If provided, use this list instead of attachment matching
149 Returns:
150 List of guardrail names to apply
151 """
152 from litellm.proxy.policy_engine.policy_matcher import PolicyMatcher
153 from litellm.proxy.policy_engine.policy_registry import get_policy_registry
155 if policies is None:
156 registry: Final = get_policy_registry()
157 if not registry.is_initialized(): 157 ↛ 158line 157 didn't jump to line 158 because the condition on line 157 was never true
158 return []
159 policies = registry.get_all_policies()
161 # Use provided policy names or get matching policies via attachments
162 matching_policy_names: Final = (
163 policy_names if policy_names is not None else PolicyMatcher.get_matching_policies(context=context)
164 )
166 if not matching_policy_names:
167 verbose_proxy_logger.debug(
168 "No policies match context: team_alias=%s, key_alias=%s, model=%s",
169 context.team_alias,
170 context.key_alias,
171 context.model,
172 )
173 return []
175 # Resolve each matching policy and combine guardrails
176 all_guardrails: Final[set[str]] = set()
178 for policy_name in matching_policy_names:
179 resolved = PolicyResolver.resolve_policy_guardrails(
180 policy_name=policy_name,
181 policies=policies,
182 context=context,
183 )
184 all_guardrails.update(resolved.guardrails)
185 verbose_proxy_logger.debug("Policy '%s' contributes guardrails: %s", policy_name, resolved.guardrails)
187 result: Final = list(all_guardrails)
188 verbose_proxy_logger.debug("Final guardrails for context: %s", result)
190 return result
192 @staticmethod
193 def resolve_pipelines_for_context(
194 context: PolicyMatchContext,
195 policies: dict[str, Policy] | None = None,
196 policy_names: list[str] | None = None,
197 ) -> list[tuple[str, GuardrailPipeline]]:
198 """
199 Resolve pipelines from matching policies for a request context.
201 Returns (policy_name, pipeline) tuples for policies that have pipelines.
202 Guardrails managed by pipelines should be excluded from the flat
203 guardrails list to avoid double execution.
205 Args:
206 context: The request context
207 policies: Dictionary of all policies (if None, uses global registry)
208 policy_names: If provided, use this list instead of attachment matching
210 Returns:
211 List of (policy_name, GuardrailPipeline) tuples
212 """
213 from litellm.proxy.policy_engine.condition_evaluator import ConditionEvaluator
214 from litellm.proxy.policy_engine.policy_matcher import PolicyMatcher
215 from litellm.proxy.policy_engine.policy_registry import get_policy_registry
217 if policies is None: 217 ↛ 223line 217 didn't jump to line 223 because the condition on line 217 was always true
218 registry: Final = get_policy_registry()
219 if not registry.is_initialized(): 219 ↛ 220line 219 didn't jump to line 220 because the condition on line 219 was never true
220 return []
221 policies = registry.get_all_policies()
223 matching_policy_names: Final = (
224 policy_names if policy_names is not None else PolicyMatcher.get_matching_policies(context=context)
225 )
226 if not matching_policy_names:
227 return []
229 pipelines: Final[list[tuple[str, GuardrailPipeline]]] = []
230 for policy_name in matching_policy_names:
231 policy = policies.get(policy_name)
232 if policy is None: 232 ↛ 233line 232 didn't jump to line 233 because the condition on line 232 was never true
233 continue
234 if policy.condition is not None and not ConditionEvaluator.evaluate( 234 ↛ 237line 234 didn't jump to line 237 because the condition on line 234 was never true
235 condition=policy.condition, context=context
236 ):
237 verbose_proxy_logger.debug("Policy '%s' condition did not match, skipping pipeline", policy_name)
238 continue
239 if policy.pipeline is not None: 239 ↛ 240line 239 didn't jump to line 240 because the condition on line 239 was never true
240 pipelines.append((policy_name, policy.pipeline))
241 verbose_proxy_logger.debug(
242 "Policy '%s' has pipeline with %s steps", policy_name, len(policy.pipeline.steps)
243 )
245 return pipelines
247 @staticmethod
248 def get_pipeline_managed_guardrails(
249 pipelines: list[tuple[str, GuardrailPipeline]],
250 ) -> set[str]:
251 """
252 Get the set of guardrail names managed by pipelines.
254 These guardrails should be excluded from normal independent execution.
255 """
256 managed: Final[set[str]] = set()
257 for _policy_name, pipeline in pipelines:
258 for step in pipeline.steps:
259 managed.add(step.guardrail)
260 return managed
262 @staticmethod
263 def get_all_resolved_policies(
264 policies: dict[str, Policy] | None = None,
265 context: PolicyMatchContext | None = None,
266 ) -> dict[str, ResolvedPolicy]:
267 """
268 Resolve all policies and return their final guardrails.
270 Useful for debugging and displaying policy configurations.
272 Args:
273 policies: Dictionary of all policies (if None, uses global registry)
274 context: Optional context for evaluating conditions
276 Returns:
277 Dictionary mapping policy names to ResolvedPolicy objects
278 """
279 from litellm.proxy.policy_engine.policy_registry import get_policy_registry
281 if policies is None: 281 ↛ 287line 281 didn't jump to line 287 because the condition on line 281 was always true
282 registry: Final = get_policy_registry()
283 if not registry.is_initialized(): 283 ↛ 284line 283 didn't jump to line 284 because the condition on line 283 was never true
284 return {}
285 policies = registry.get_all_policies()
287 resolved: Final[dict[str, ResolvedPolicy]] = {}
289 for policy_name in policies:
290 resolved[policy_name] = PolicyResolver.resolve_policy_guardrails(
291 policy_name=policy_name,
292 policies=policies,
293 context=context,
294 )
296 return resolved