scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-21 00:21:49 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 74

fix policy resolver

Browse source
This commit is contained in:
Ishaan Jaffer 2026-01-22 17:31:18 -08:00
parent b27c6e13fb
commit 1ca67d101c
6 changed files with 112 additions and 360 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

19
litellm/proxy/policy_engine/__init__.py
View file

@ -6,7 +6,7 @@ with scoping rules. Policies can target specific teams, API keys, and models usi
wildcard patterns, and support inheritance from base policies.
Configuration structure:
- `policies`: Define WHAT guardrails to apply (with inheritance and statements)
- `policies`: Define WHAT guardrails to apply (with inheritance and conditions)
- `policy_attachments`: Define WHERE policies apply (teams, keys, models)
Example:
@ -17,22 +17,19 @@ policies:
guardrails:
add: [pii_blocker]
healthcare-compliance:
gpt4-safety:
inherit: global-baseline
description: "Extra safety for GPT-4"
guardrails:
add: [hipaa_audit]
statements:
- sid: "GPT4Only"
guardrails: [toxicity_filter]
condition:
model:
in: ["gpt-4", "gpt-4-turbo"]
add: [toxicity_filter]
condition:
model: "gpt-4.*" # regex pattern
policy_attachments:
- policy: global-baseline
scope: "*"
- policy: healthcare-compliance
teams: [healthcare-team]
- policy: gpt4-safety
scope: "*"
```
"""

210
litellm/proxy/policy_engine/condition_evaluator.py
View file

@ -1,15 +1,14 @@
"""
Condition Evaluator - Evaluates AWS IAM-style conditions.
Condition Evaluator - Evaluates policy conditions.
Supports operators like equals, in, prefix, not_equals, not_in for
fine-grained policy statement matching.
Supports model-based conditions with exact match or regex patterns.
"""
from typing import Any, Dict, Optional
import re
from typing import List, Optional, Union
from litellm._logging import verbose_proxy_logger
from litellm.types.proxy.policy_engine import (
ConditionOperator,
PolicyCondition,
PolicyMatchContext,
)
@ -19,21 +18,16 @@ class ConditionEvaluator:
"""
Evaluates policy conditions against request context.
Supports AWS IAM-style condition operators:
- equals: Exact string match
- in: Value must be in the list
- prefix: Value must start with the prefix
- not_equals: Value must NOT equal
- not_in: Value must NOT be in the list
All conditions in a PolicyCondition must match (AND logic).
Supports model conditions with:
- Exact string match: "gpt-4"
- Regex pattern: "gpt-4.*"
- List of values: ["gpt-4", "gpt-4-turbo"]
"""
@staticmethod
def evaluate(
condition: Optional[PolicyCondition],
context: PolicyMatchContext,
metadata: Optional[Dict[str, Any]] = None,
) -> bool:
"""
Evaluate a policy condition against a request context.
@ -41,7 +35,6 @@ class ConditionEvaluator:
Args:
condition: The condition to evaluate (None = always matches)
context: The request context with team, key, model
metadata: Optional request metadata for metadata conditions
Returns:
True if condition matches, False otherwise
@ -52,166 +45,67 @@ class ConditionEvaluator:
# Check model condition
if condition.model is not None:
if not ConditionEvaluator.evaluate_operator(
operator=condition.model,
value=context.model,
if not ConditionEvaluator._evaluate_model_condition(
condition=condition.model,
model=context.model,
):
verbose_proxy_logger.debug(
f"Condition failed: model={context.model} did not match {condition.model}"
)
return False
# Check team condition
if condition.team is not None:
if not ConditionEvaluator.evaluate_operator(
operator=condition.team,
value=context.team_alias,
):
verbose_proxy_logger.debug(
f"Condition failed: team={context.team_alias} did not match {condition.team}"
)
return False
# Check key condition
if condition.key is not None:
if not ConditionEvaluator.evaluate_operator(
operator=condition.key,
value=context.key_alias,
):
verbose_proxy_logger.debug(
f"Condition failed: key={context.key_alias} did not match {condition.key}"
)
return False
# Check metadata conditions
if condition.metadata is not None and metadata is not None:
for field_name, field_operator in condition.metadata.items():
field_value = metadata.get(field_name)
# Convert to string for comparison
field_value_str = str(field_value) if field_value is not None else None
if not ConditionEvaluator.evaluate_operator(
operator=field_operator,
value=field_value_str,
):
verbose_proxy_logger.debug(
f"Condition failed: metadata.{field_name}={field_value} "
f"did not match {field_operator}"
)
return False
return True
@staticmethod
def evaluate_operator(
operator: ConditionOperator,
value: Optional[str],
def _evaluate_model_condition(
condition: Union[str, List[str]],
model: Optional[str],
) -> bool:
"""
Evaluate a single condition operator against a value.
Evaluate a model condition.
Args:
operator: The condition operator to evaluate
value: The value to check (None if not provided)
condition: String (exact or regex) or list of strings
model: The model name to check
Returns:
True if the value matches the operator, False otherwise
True if model matches condition, False otherwise
"""
# Handle None value
if value is None:
# For positive operators (equals, in, prefix), None never matches
if operator.equals is not None:
return False
if operator.in_ is not None:
return False
if operator.prefix is not None:
return False
# For negative operators (not_equals, not_in), None always matches
# (None is not equal to anything, and not in any list)
if operator.not_equals is not None:
return True
if operator.not_in is not None:
return True
# No operators specified = matches
if model is None:
return False
# Handle list of values
if isinstance(condition, list):
return any(
ConditionEvaluator._matches_pattern(pattern, model)
for pattern in condition
)
# Single value - check as pattern
return ConditionEvaluator._matches_pattern(condition, model)
@staticmethod
def _matches_pattern(pattern: str, value: str) -> bool:
"""
Check if value matches pattern (exact match or regex).
Args:
pattern: Pattern to match (exact string or regex)
value: Value to check
Returns:
True if matches, False otherwise
"""
# First try exact match
if pattern == value:
return True
# Check equals
if operator.equals is not None:
if value != operator.equals:
return False
# Check in (value must be in list)
if operator.in_ is not None:
if value not in operator.in_:
return False
# Check prefix
if operator.prefix is not None:
if not value.startswith(operator.prefix):
return False
# Check not_equals
if operator.not_equals is not None:
if value == operator.not_equals:
return False
# Check not_in
if operator.not_in is not None:
if value in operator.not_in:
return False
return True
@staticmethod
def evaluate_all_conditions(
conditions: list,
context: PolicyMatchContext,
metadata: Optional[Dict[str, Any]] = None,
) -> bool:
"""
Evaluate multiple conditions (AND logic - all must match).
Args:
conditions: List of PolicyCondition objects
context: The request context
metadata: Optional request metadata
Returns:
True if ALL conditions match, False otherwise
"""
for condition in conditions:
if not ConditionEvaluator.evaluate(
condition=condition,
context=context,
metadata=metadata,
):
return False
return True
@staticmethod
def evaluate_any_condition(
conditions: list,
context: PolicyMatchContext,
metadata: Optional[Dict[str, Any]] = None,
) -> bool:
"""
Evaluate multiple conditions (OR logic - any must match).
Args:
conditions: List of PolicyCondition objects
context: The request context
metadata: Optional request metadata
Returns:
True if ANY condition matches, False otherwise
"""
if not conditions:
return True
for condition in conditions:
if ConditionEvaluator.evaluate(
condition=condition,
context=context,
metadata=metadata,
):
# Try as regex pattern
try:
if re.fullmatch(pattern, value):
return True
except re.error:
# Invalid regex, treat as literal string (already checked above)
pass
return False

100
litellm/proxy/policy_engine/policy_registry.py
View file

@ -11,12 +11,10 @@ from typing import Any, Dict, List, Optional
from litellm._logging import verbose_proxy_logger
from litellm.types.proxy.policy_engine import (
ConditionOperator,
Policy,
PolicyCondition,
PolicyConfig,
PolicyGuardrails,
PolicyStatement,
)
@ -30,7 +28,7 @@ class PolicyRegistry:
Policies define WHAT guardrails to apply:
- Base guardrails via guardrails.add/remove
- Inheritance via inherit field
- Conditional guardrails via statements
- Conditional guardrails via condition.model
"""
def __init__(self):
@ -83,103 +81,19 @@ class PolicyRegistry:
# Handle legacy format where guardrails might be a list
guardrails = PolicyGuardrails(add=guardrails_data if guardrails_data else None)
# Parse statements (conditional guardrails)
statements = None
statements_data = policy_data.get("statements")
if statements_data:
statements = [
self._parse_statement(stmt_data) for stmt_data in statements_data
]
# Parse condition (simple model-based condition)
condition = None
condition_data = policy_data.get("condition")
if condition_data:
condition = PolicyCondition(model=condition_data.get("model"))
return Policy(
inherit=policy_data.get("inherit"),
guardrails=guardrails,
statements=statements,
description=policy_data.get("description"),
)
def _parse_statement(self, stmt_data: Dict[str, Any]) -> PolicyStatement:
"""
Parse a policy statement from raw configuration data.
Args:
stmt_data: Raw statement configuration
Returns:
Parsed PolicyStatement object
"""
condition = None
condition_data = stmt_data.get("condition")
if condition_data:
condition = self._parse_condition(condition_data)
return PolicyStatement(
sid=stmt_data.get("sid"),
guardrails=stmt_data.get("guardrails", []),
guardrails=guardrails,
condition=condition,
)
def _parse_condition(self, condition_data: Dict[str, Any]) -> PolicyCondition:
"""
Parse a policy condition from raw configuration data.
Args:
condition_data: Raw condition configuration
Returns:
Parsed PolicyCondition object
"""
model_op = None
team_op = None
key_op = None
metadata_ops = None
if "model" in condition_data:
model_op = self._parse_operator(condition_data["model"])
if "team" in condition_data:
team_op = self._parse_operator(condition_data["team"])
if "key" in condition_data:
key_op = self._parse_operator(condition_data["key"])
if "metadata" in condition_data:
metadata_ops = {
field: self._parse_operator(op_data)
for field, op_data in condition_data["metadata"].items()
}
return PolicyCondition(
model=model_op,
team=team_op,
key=key_op,
metadata=metadata_ops,
)
def _parse_operator(self, op_data: Dict[str, Any]) -> ConditionOperator:
"""
Parse a condition operator from raw configuration data.
Args:
op_data: Raw operator configuration
Returns:
Parsed ConditionOperator object
"""
# Build dict with alias names for Pydantic
operator_dict: Dict[str, Any] = {}
if "equals" in op_data:
operator_dict["equals"] = op_data["equals"]
if "in" in op_data:
operator_dict["in"] = op_data["in"]
if "prefix" in op_data:
operator_dict["prefix"] = op_data["prefix"]
if "not_equals" in op_data:
operator_dict["not_equals"] = op_data["not_equals"]
if "notIn" in op_data:
operator_dict["notIn"] = op_data["notIn"]
elif "not_in" in op_data:
operator_dict["notIn"] = op_data["not_in"]
return ConditionOperator.model_validate(operator_dict)
def get_policy(self, policy_name: str) -> Optional[Policy]:
"""
Get a policy by name.

82
litellm/proxy/policy_engine/policy_resolver.py
View file

@ -4,7 +4,7 @@ Policy Resolver - Resolves final guardrail list from policies.
Handles:
- Inheritance chain resolution (inherit with add/remove)
- Applying add/remove guardrails
- Evaluating conditional statements
- Evaluating model conditions
- Combining guardrails from multiple matching policies
"""
@ -24,7 +24,7 @@ class PolicyResolver:
Handles:
- Inheritance chains with add/remove operations
- Conditional statements with AWS IAM-style conditions
- Model-based conditions
"""
@staticmethod
@ -72,7 +72,6 @@ class PolicyResolver:
policy_name: str,
policies: Dict[str, Policy],
context: Optional[PolicyMatchContext] = None,
metadata: Optional[Dict[str, Any]] = None,
) -> ResolvedPolicy:
"""
Resolve the final guardrails for a single policy, including inheritance.
@ -80,17 +79,18 @@ class PolicyResolver:
This method:
1. Resolves the inheritance chain
2. Applies add/remove from each policy in the chain
3. Evaluates conditional statements (if context provided)
3. Evaluates model conditions (if context provided)
Args:
policy_name: Name of the policy to resolve
policies: Dictionary of all policies
context: Optional request context for evaluating statement conditions
metadata: Optional request metadata for condition evaluation
context: Optional request context for evaluating conditions
Returns:
ResolvedPolicy with final guardrails list
"""
from litellm.proxy.policy_engine.condition_evaluator import ConditionEvaluator
inheritance_chain = PolicyResolver.resolve_inheritance_chain(
policy_name=policy_name, policies=policies
)
@ -104,6 +104,17 @@ class PolicyResolver:
if policy is None:
continue
# Check if policy condition matches (if context provided)
if context is not None and policy.condition is not None:
if not ConditionEvaluator.evaluate(
condition=policy.condition,
context=context,
):
verbose_proxy_logger.debug(
f"Policy '{chain_policy_name}' condition did not match, skipping guardrails"
)
continue
# Add guardrails from guardrails.add
for guardrail in policy.guardrails.get_add():
guardrails.add(guardrail)
@ -112,66 +123,16 @@ class PolicyResolver:
for guardrail in policy.guardrails.get_remove():
guardrails.discard(guardrail)
# Evaluate statements (if context provided)
if context is not None and policy.statements:
statement_guardrails = PolicyResolver._evaluate_statements(
statements=policy.statements,
context=context,
metadata=metadata,
)
guardrails.update(statement_guardrails)
if statement_guardrails:
verbose_proxy_logger.debug(
f"Policy '{chain_policy_name}' statements contributed: {statement_guardrails}"
)
return ResolvedPolicy(
policy_name=policy_name,
guardrails=list(guardrails),
inheritance_chain=inheritance_chain,
)
@staticmethod
def _evaluate_statements(
statements: list,
context: PolicyMatchContext,
metadata: Optional[Dict[str, Any]] = None,
) -> Set[str]:
"""
Evaluate policy statements and return guardrails from matching statements.
Args:
statements: List of PolicyStatement objects
context: The request context
metadata: Optional request metadata
Returns:
Set of guardrail names from matching statements
"""
from litellm.proxy.policy_engine.condition_evaluator import ConditionEvaluator
guardrails: Set[str] = set()
for statement in statements:
# Evaluate the statement's condition
if ConditionEvaluator.evaluate(
condition=statement.condition,
context=context,
metadata=metadata,
):
guardrails.update(statement.guardrails)
verbose_proxy_logger.debug(
f"Statement '{statement.sid or 'unnamed'}' matched, "
f"adding guardrails: {statement.guardrails}"
)
return guardrails
@staticmethod
def resolve_guardrails_for_context(
context: PolicyMatchContext,
policies: Optional[Dict[str, Policy]] = None,
metadata: Optional[Dict[str, Any]] = None,
) -> List[str]:
"""
Resolve the final list of guardrails for a request context.
@ -179,13 +140,12 @@ class PolicyResolver:
This:
1. Finds all policies that match the context via policy_attachments
2. Resolves each policy's guardrails (including inheritance)
3. Evaluates conditional statements
3. Evaluates model conditions
4. Combines all guardrails (union)
Args:
context: The request context
policies: Dictionary of all policies (if None, uses global registry)
metadata: Optional request metadata for condition evaluation
Returns:
List of guardrail names to apply
@ -217,7 +177,6 @@ class PolicyResolver:
policy_name=policy_name,
policies=policies,
context=context,
metadata=metadata,
)
all_guardrails.update(resolved.guardrails)
verbose_proxy_logger.debug(
@ -235,7 +194,6 @@ class PolicyResolver:
def get_all_resolved_policies(
policies: Optional[Dict[str, Policy]] = None,
context: Optional[PolicyMatchContext] = None,
metadata: Optional[Dict[str, Any]] = None,
) -> Dict[str, ResolvedPolicy]:
"""
Resolve all policies and return their final guardrails.
@ -244,8 +202,7 @@ class PolicyResolver:
Args:
policies: Dictionary of all policies (if None, uses global registry)
context: Optional context for evaluating statement conditions
metadata: Optional metadata for condition evaluation
context: Optional context for evaluating conditions
Returns:
Dictionary mapping policy names to ResolvedPolicy objects
@ -265,7 +222,6 @@ class PolicyResolver:
policy_name=policy_name,
policies=policies,
context=context,
metadata=metadata,
)
return resolved

14
litellm/types/proxy/policy_engine/__init__.py
View file

@ -5,23 +5,18 @@ The Policy Engine allows administrators to define policies that combine guardrai
with scoping rules. Policies can target specific teams, API keys, and models using
wildcard patterns, and support inheritance from base policies.
The engine supports two configuration styles:
1. **Simple Style**: Policies with inline scope (original)
2. **Advanced Style**: Policies with statements and separate attachments
Both styles support the `inherit` field with `guardrails.add` and `guardrails.remove`.
Configuration:
- `policies`: Define WHAT guardrails to apply (with inheritance and conditions)
- `policy_attachments`: Define WHERE policies apply (teams, keys, models)
"""
from litellm.types.proxy.policy_engine.policy_types import (
ConditionOperator,
Policy,
PolicyAttachment,
PolicyCondition,
PolicyConfig,
PolicyGuardrails,
PolicyScope,
PolicyStatement,
)
from litellm.types.proxy.policy_engine.resolver_types import (
PolicyGuardrailsResponse,
@ -46,10 +41,7 @@ __all__ = [
"PolicyConfig",
"PolicyGuardrails",
"PolicyScope",
# Condition types (new)
"ConditionOperator",
"PolicyCondition",
"PolicyStatement",
"PolicyAttachment",
# Validation types
"PolicyValidateRequest",

47
proxy_config.yaml
View file

@ -20,9 +20,10 @@ general_settings:
#
# Policies define guardrails with:
# - inherit: Inherit guardrails from another policy
# - description: Human-readable description
# - guardrails.add: Add guardrails (on top of inherited)
# - guardrails.remove: Remove guardrails (from inherited)
# - statements: Conditional guardrails with AWS IAM-style conditions
# - condition.model: Model pattern (exact or regex) for when guardrails apply
#
policies:
# Global baseline policy
@ -50,25 +51,23 @@ policies:
remove:
- pii_blocker
# Policy with conditional statements
conditional-safety:
description: "Model-specific guardrails using conditions"
# Policy with model condition (regex pattern)
gpt4-safety:
description: "Extra safety for GPT-4 models"
guardrails:
add:
- base_safety
statements:
- sid: "GPT4ToxicityFilter"
guardrails:
- toxicity_filter
condition:
model:
in: ["gpt-4", "gpt-4-turbo", "gpt-4o"]
- sid: "BedrockPIICheck"
guardrails:
- strict_pii_blocker
condition:
model:
prefix: "bedrock/"
- toxicity_filter
condition:
model: "gpt-4.*" # regex: matches gpt-4, gpt-4-turbo, gpt-4o, etc.
# Policy with model condition (exact match list)
bedrock-compliance:
description: "Compliance for Bedrock models"
guardrails:
add:
- strict_pii_blocker
condition:
model: ["bedrock/claude-3", "bedrock/claude-2"] # exact matches
# ───────────────────────────────────────────────
# POLICY ATTACHMENTS - Define WHERE policies apply
@ -94,9 +93,9 @@ policy_attachments:
- "dev-key-*"
- "test-key-*"
# Model-specific attachment
- policy: conditional-safety
models:
- "gpt-4"
- "gpt-4-turbo"
- "bedrock/*"
# Model-specific policies (attached globally, condition filters by model)
- policy: gpt4-safety
scope: "*"
- policy: bedrock-compliance
scope: "*"