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

new attachment config

Browse source
This commit is contained in:
Ishaan Jaffer 2026-01-22 17:26:58 -08:00
parent c0aef7e16f
commit 04a93a6efb
12 changed files with 1208 additions and 84 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

28
litellm/proxy/litellm_pre_call_utils.py
View file

@ -1377,6 +1377,7 @@ def add_guardrails_from_policy_engine(
metadata_variable_name: The name of the metadata field in data
user_api_key_dict: The user's API key authentication info
"""
from litellm._logging import verbose_proxy_logger
from litellm.proxy.common_utils.callback_utils import (
add_policy_to_applied_policies_header,
)
@ -1386,7 +1387,12 @@ def add_guardrails_from_policy_engine(
from litellm.types.proxy.policy_engine import PolicyMatchContext
registry = get_policy_registry()
verbose_proxy_logger.debug(
f"Policy engine: registry initialized={registry.is_initialized()}, "
f"policy_count={len(registry.get_all_policies())}"
)
if not registry.is_initialized():
verbose_proxy_logger.debug("Policy engine not initialized, skipping policy matching")
return
# Build context from request
@ -1396,12 +1402,16 @@ def add_guardrails_from_policy_engine(
model=data.get("model"),
)
# Get matching policies
policies = registry.get_all_policies()
matching_policy_names = PolicyMatcher.get_matching_policies(
policies=policies, context=context
verbose_proxy_logger.debug(
f"Policy engine: matching policies for context team_alias={context.team_alias}, "
f"key_alias={context.key_alias}, model={context.model}"
)
# Get matching policies via attachments
matching_policy_names = PolicyMatcher.get_matching_policies(context=context)
verbose_proxy_logger.debug(f"Policy engine: matched policies: {matching_policy_names}")
if not matching_policy_names:
return
@ -1412,9 +1422,9 @@ def add_guardrails_from_policy_engine(
)
# Resolve guardrails from matching policies
resolved_guardrails = PolicyResolver.resolve_guardrails_for_context(
context=context, policies=policies
)
resolved_guardrails = PolicyResolver.resolve_guardrails_for_context(context=context)
verbose_proxy_logger.debug(f"Policy engine: resolved guardrails: {resolved_guardrails}")
if not resolved_guardrails:
return
@ -1432,6 +1442,10 @@ def add_guardrails_from_policy_engine(
combined.update(resolved_guardrails)
data[metadata_variable_name]["guardrails"] = list(combined)
verbose_proxy_logger.debug(
f"Policy engine: added guardrails to request metadata: {list(combined)}"
)
def add_provider_specific_headers_to_request(
data: dict,

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

@ -4,8 +4,43 @@ LiteLLM Policy Engine
The Policy Engine allows administrators to define policies that combine guardrails
with scoping rules. Policies can target specific teams, API keys, and models using
wildcard patterns, and support inheritance from base policies.
Configuration structure:
- `policies`: Define WHAT guardrails to apply (with inheritance and statements)
- `policy_attachments`: Define WHERE policies apply (teams, keys, models)
Example:
```yaml
policies:
global-baseline:
description: "Base guardrails for all requests"
guardrails:
add: [pii_blocker]
healthcare-compliance:
inherit: global-baseline
guardrails:
add: [hipaa_audit]
statements:
- sid: "GPT4Only"
guardrails: [toxicity_filter]
condition:
model:
in: ["gpt-4", "gpt-4-turbo"]
policy_attachments:
- policy: global-baseline
scope: "*"
- policy: healthcare-compliance
teams: [healthcare-team]
```
"""
from litellm.proxy.policy_engine.attachment_registry import (
AttachmentRegistry,
get_attachment_registry,
)
from litellm.proxy.policy_engine.condition_evaluator import ConditionEvaluator
from litellm.proxy.policy_engine.policy_matcher import PolicyMatcher
from litellm.proxy.policy_engine.policy_registry import (
PolicyRegistry,
@ -15,9 +50,14 @@ from litellm.proxy.policy_engine.policy_resolver import PolicyResolver
from litellm.proxy.policy_engine.policy_validator import PolicyValidator
__all__ = [
# Registries
"PolicyRegistry",
"get_policy_registry",
"AttachmentRegistry",
"get_attachment_registry",
# Core components
"PolicyMatcher",
"PolicyResolver",
"PolicyValidator",
"ConditionEvaluator",
]

207
litellm/proxy/policy_engine/attachment_registry.py Normal file
View file

@ -0,0 +1,207 @@
"""
Attachment Registry - Manages policy attachments from YAML config.
Attachments define WHERE policies apply, separate from the policy definitions.
This allows the same policy to be attached to multiple scopes.
"""
from typing import Any, Dict, List, Optional
from litellm._logging import verbose_proxy_logger
from litellm.types.proxy.policy_engine import (
PolicyAttachment,
PolicyMatchContext,
PolicyScope,
)
class AttachmentRegistry:
"""
In-memory registry for storing and managing policy attachments.
Attachments define the relationship between policies and their scopes.
A single policy can have multiple attachments (applied to different scopes).
Example YAML:
```yaml
attachments:
- policy: global-baseline
scope: "*"
- policy: healthcare-compliance
teams: [healthcare-team]
- policy: dev-safety
keys: ["dev-key-*"]
```
"""
def __init__(self):
self._attachments: List[PolicyAttachment] = []
self._initialized: bool = False
def load_attachments(self, attachments_config: List[Dict[str, Any]]) -> None:
"""
Load attachments from a configuration list.
Args:
attachments_config: List of attachment dictionaries from YAML.
"""
self._attachments = []
for attachment_data in attachments_config:
try:
attachment = self._parse_attachment(attachment_data)
self._attachments.append(attachment)
verbose_proxy_logger.debug(
f"Loaded attachment for policy: {attachment.policy}"
)
except Exception as e:
verbose_proxy_logger.error(
f"Error loading attachment: {str(e)}"
)
raise ValueError(f"Invalid attachment: {str(e)}") from e
self._initialized = True
verbose_proxy_logger.info(f"Loaded {len(self._attachments)} policy attachments")
def _parse_attachment(self, attachment_data: Dict[str, Any]) -> PolicyAttachment:
"""
Parse an attachment from raw configuration data.
Args:
attachment_data: Raw attachment configuration
Returns:
Parsed PolicyAttachment object
"""
return PolicyAttachment(
policy=attachment_data.get("policy", ""),
scope=attachment_data.get("scope"),
teams=attachment_data.get("teams"),
keys=attachment_data.get("keys"),
models=attachment_data.get("models"),
)
def get_attached_policies(self, context: PolicyMatchContext) -> List[str]:
"""
Get list of policy names attached to the given context.
Args:
context: The request context to match against
Returns:
List of policy names that are attached to matching scopes
"""
from litellm.proxy.policy_engine.policy_matcher import PolicyMatcher
attached_policies: List[str] = []
for attachment in self._attachments:
scope = attachment.to_policy_scope()
if PolicyMatcher.scope_matches(scope=scope, context=context):
if attachment.policy not in attached_policies:
attached_policies.append(attachment.policy)
verbose_proxy_logger.debug(
f"Attachment matched: policy={attachment.policy}, "
f"context=(team={context.team_alias}, key={context.key_alias}, model={context.model})"
)
return attached_policies
def is_policy_attached(
self, policy_name: str, context: PolicyMatchContext
) -> bool:
"""
Check if a specific policy is attached to the given context.
Args:
policy_name: Name of the policy to check
context: The request context to match against
Returns:
True if the policy is attached to a matching scope
"""
attached = self.get_attached_policies(context)
return policy_name in attached
def get_all_attachments(self) -> List[PolicyAttachment]:
"""
Get all loaded attachments.
Returns:
List of all PolicyAttachment objects
"""
return self._attachments.copy()
def get_attachments_for_policy(self, policy_name: str) -> List[PolicyAttachment]:
"""
Get all attachments for a specific policy.
Args:
policy_name: Name of the policy
Returns:
List of attachments for the policy
"""
return [a for a in self._attachments if a.policy == policy_name]
def is_initialized(self) -> bool:
"""
Check if the registry has been initialized with attachments.
Returns:
True if attachments have been loaded, False otherwise
"""
return self._initialized
def clear(self) -> None:
"""
Clear all attachments from the registry.
"""
self._attachments = []
self._initialized = False
def add_attachment(self, attachment: PolicyAttachment) -> None:
"""
Add a single attachment.
Args:
attachment: PolicyAttachment object to add
"""
self._attachments.append(attachment)
verbose_proxy_logger.debug(f"Added attachment for policy: {attachment.policy}")
def remove_attachments_for_policy(self, policy_name: str) -> int:
"""
Remove all attachments for a specific policy.
Args:
policy_name: Name of the policy
Returns:
Number of attachments removed
"""
original_count = len(self._attachments)
self._attachments = [a for a in self._attachments if a.policy != policy_name]
removed_count = original_count - len(self._attachments)
if removed_count > 0:
verbose_proxy_logger.debug(
f"Removed {removed_count} attachment(s) for policy: {policy_name}"
)
return removed_count
# Global singleton instance
_attachment_registry: Optional[AttachmentRegistry] = None
def get_attachment_registry() -> AttachmentRegistry:
"""
Get the global AttachmentRegistry singleton.
Returns:
The global AttachmentRegistry instance
"""
global _attachment_registry
if _attachment_registry is None:
_attachment_registry = AttachmentRegistry()
return _attachment_registry

217
litellm/proxy/policy_engine/condition_evaluator.py Normal file
View file

@ -0,0 +1,217 @@
"""
Condition Evaluator - Evaluates AWS IAM-style conditions.
Supports operators like equals, in, prefix, not_equals, not_in for
fine-grained policy statement matching.
"""
from typing import Any, Dict, Optional
from litellm._logging import verbose_proxy_logger
from litellm.types.proxy.policy_engine import (
ConditionOperator,
PolicyCondition,
PolicyMatchContext,
)
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).
"""
@staticmethod
def evaluate(
condition: Optional[PolicyCondition],
context: PolicyMatchContext,
metadata: Optional[Dict[str, Any]] = None,
) -> bool:
"""
Evaluate a policy condition against a request context.
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
"""
# No condition means always matches
if condition is None:
return True
# Check model condition
if condition.model is not None:
if not ConditionEvaluator.evaluate_operator(
operator=condition.model,
value=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],
) -> bool:
"""
Evaluate a single condition operator against a value.
Args:
operator: The condition operator to evaluate
value: The value to check (None if not provided)
Returns:
True if the value matches the operator, 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
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,
):
return True
return False

137
litellm/proxy/policy_engine/init_policies.py
View file

@ -1,10 +1,15 @@
"""
Policy Initialization - Loads policies from config and validates on startup.
Configuration structure:
- policies: Define WHAT guardrails to apply (with inheritance and statements)
- policy_attachments: Define WHERE policies apply (teams, keys, models)
"""
from typing import TYPE_CHECKING, Any, Dict, List, Optional
from litellm._logging import verbose_proxy_logger
from litellm.proxy.policy_engine.attachment_registry import get_attachment_registry
from litellm.proxy.policy_engine.policy_registry import get_policy_registry
from litellm.proxy.policy_engine.policy_validator import PolicyValidator
from litellm.types.proxy.policy_engine import PolicyValidationResponse
@ -12,9 +17,85 @@ from litellm.types.proxy.policy_engine import PolicyValidationResponse
if TYPE_CHECKING:
from litellm.proxy.utils import PrismaClient
# ANSI color codes for terminal output
_green_color_code = "\033[92m"
_blue_color_code = "\033[94m"
_yellow_color_code = "\033[93m"
_reset_color_code = "\033[0m"
def _print_policies_on_startup(
policies_config: Dict[str, Any],
policy_attachments_config: Optional[List[Dict[str, Any]]] = None,
) -> None:
"""
Print loaded policies to console on startup (similar to model list).
"""
import sys
print( # noqa: T201
f"{_green_color_code}\nLiteLLM Policy Engine: Loaded {len(policies_config)} policies{_reset_color_code}\n"
)
sys.stdout.flush()
for policy_name, policy_data in policies_config.items():
guardrails = policy_data.get("guardrails", {})
inherit = policy_data.get("inherit")
statements = policy_data.get("statements", [])
description = policy_data.get("description")
guardrails_add = guardrails.get("add", []) if isinstance(guardrails, dict) else []
guardrails_remove = guardrails.get("remove", []) if isinstance(guardrails, dict) else []
inherit_str = f" (inherits: {inherit})" if inherit else ""
print( # noqa: T201
f"{_blue_color_code} - {policy_name}{inherit_str}{_reset_color_code}"
)
if description:
print(f" description: {description}") # noqa: T201
if guardrails_add:
print(f" guardrails.add: {guardrails_add}") # noqa: T201
if guardrails_remove:
print(f" guardrails.remove: {guardrails_remove}") # noqa: T201
if statements:
print(f" statements: {len(statements)} conditional statement(s)") # noqa: T201
# Print attachments
if policy_attachments_config:
print( # noqa: T201
f"\n{_yellow_color_code}Policy Attachments: {len(policy_attachments_config)} attachment(s){_reset_color_code}"
)
for attachment in policy_attachments_config:
policy = attachment.get("policy", "unknown")
scope = attachment.get("scope")
teams = attachment.get("teams")
keys = attachment.get("keys")
models = attachment.get("models")
scope_parts = []
if scope == "*":
scope_parts.append("scope=* (global)")
if teams:
scope_parts.append(f"teams={teams}")
if keys:
scope_parts.append(f"keys={keys}")
if models:
scope_parts.append(f"models={models}")
scope_str = ", ".join(scope_parts) if scope_parts else "all"
print(f" - {policy} -> {scope_str}") # noqa: T201
else:
print( # noqa: T201
f"\n{_yellow_color_code}Warning: No policy_attachments configured. Policies will not be applied to any requests.{_reset_color_code}"
)
print() # noqa: T201
sys.stdout.flush()
async def init_policies(
policies_config: Dict[str, Any],
policy_attachments_config: Optional[List[Dict[str, Any]]] = None,
prisma_client: Optional["PrismaClient"] = None,
validate_db: bool = True,
fail_on_error: bool = True,
@ -26,9 +107,11 @@ async def init_policies(
1. Parses the policy configuration
2. Validates policies (guardrails exist, teams/keys exist in DB)
3. Loads policies into the global registry
4. Loads attachments into the attachment registry (if provided)
Args:
policies_config: Dictionary mapping policy names to policy definitions
policy_attachments_config: Optional list of policy attachment configurations
prisma_client: Optional Prisma client for database validation
validate_db: Whether to validate team/key aliases against database
fail_on_error: If True, raise exception on validation errors
@ -41,8 +124,12 @@ async def init_policies(
"""
verbose_proxy_logger.info(f"Initializing {len(policies_config)} policies...")
# Get the global registry
registry = get_policy_registry()
# Print policies to console on startup
_print_policies_on_startup(policies_config, policy_attachments_config)
# Get the global registries
policy_registry = get_policy_registry()
attachment_registry = get_attachment_registry()
# Create validator
validator = PolicyValidator(prisma_client=prisma_client)
@ -80,7 +167,7 @@ async def init_policies(
# Load policies into registry (even with warnings)
try:
registry.load_policies(policies_config)
policy_registry.load_policies(policies_config)
verbose_proxy_logger.info(
f"Successfully loaded {len(policies_config)} policies"
)
@ -88,11 +175,23 @@ async def init_policies(
verbose_proxy_logger.error(f"Failed to load policies: {str(e)}")
raise
# Load attachments if provided
if policy_attachments_config:
try:
attachment_registry.load_attachments(policy_attachments_config)
verbose_proxy_logger.info(
f"Successfully loaded {len(policy_attachments_config)} policy attachments"
)
except Exception as e:
verbose_proxy_logger.error(f"Failed to load policy attachments: {str(e)}")
raise
return validation_result
def init_policies_sync(
policies_config: Dict[str, Any],
policy_attachments_config: Optional[List[Dict[str, Any]]] = None,
fail_on_error: bool = True,
) -> None:
"""
@ -102,6 +201,7 @@ def init_policies_sync(
Args:
policies_config: Dictionary mapping policy names to policy definitions
policy_attachments_config: Optional list of policy attachment configurations
fail_on_error: If True, raise exception on validation errors
"""
import asyncio
@ -116,6 +216,7 @@ def init_policies_sync(
loop.run_until_complete(
init_policies(
policies_config=policies_config,
policy_attachments_config=policy_attachments_config,
prisma_client=None,
validate_db=False,
fail_on_error=fail_on_error,
@ -132,32 +233,42 @@ def get_policies_summary() -> Dict[str, Any]:
"""
from litellm.proxy.policy_engine.policy_resolver import PolicyResolver
registry = get_policy_registry()
policy_registry = get_policy_registry()
attachment_registry = get_attachment_registry()
if not registry.is_initialized():
return {"initialized": False, "policies": {}}
if not policy_registry.is_initialized():
return {"initialized": False, "policies": {}, "attachments": []}
resolved = PolicyResolver.get_all_resolved_policies()
summary = {
summary: Dict[str, Any] = {
"initialized": True,
"policy_count": len(resolved),
"attachment_count": len(attachment_registry.get_all_attachments()),
"policies": {},
"attachments": [],
}
for policy_name, resolved_policy in resolved.items():
policy = registry.get_policy(policy_name)
policy = policy_registry.get_policy(policy_name)
summary["policies"][policy_name] = {
"inherit": policy.inherit if policy else None,
"scope": {
"teams": policy.scope.get_teams() if policy else [],
"keys": policy.scope.get_keys() if policy else [],
"models": policy.scope.get_models() if policy else [],
},
"description": policy.description if policy else None,
"guardrails_add": policy.guardrails.get_add() if policy else [],
"guardrails_remove": policy.guardrails.get_remove() if policy else [],
"statements_count": len(policy.statements) if policy and policy.statements else 0,
"resolved_guardrails": resolved_policy.guardrails,
"inheritance_chain": resolved_policy.inheritance_chain,
}
# Add attachment info
for attachment in attachment_registry.get_all_attachments():
summary["attachments"].append({
"policy": attachment.policy,
"scope": attachment.scope,
"teams": attachment.teams,
"keys": attachment.keys,
"models": attachment.models,
})
return summary

44
litellm/proxy/policy_engine/policy_matcher.py
View file

@ -1,8 +1,10 @@
"""
Policy Matcher - Matches requests against policy scopes.
Policy Matcher - Matches requests against policy attachments.
Uses existing wildcard pattern matching helpers to determine which policies
apply to a given request based on team alias, key alias, and model.
Policies are matched via policy_attachments which define WHERE each policy applies.
"""
from typing import Dict, List, Optional
@ -14,11 +16,13 @@ from litellm.types.proxy.policy_engine import Policy, PolicyMatchContext, Policy
class PolicyMatcher:
"""
Matches incoming requests against policy scopes.
Matches incoming requests against policy attachments.
Supports wildcard patterns:
- "*" matches everything
- "prefix-*" matches anything starting with "prefix-"
Uses policy_attachments to determine which policies apply to a request.
"""
@staticmethod
@ -81,32 +85,29 @@ class PolicyMatcher:
@staticmethod
def get_matching_policies(
policies: Dict[str, Policy],
context: PolicyMatchContext,
) -> List[str]:
"""
Get list of policy names that match the given context.
Get list of policy names that match the given context via attachments.
Args:
policies: Dictionary of all policies
context: The request context to match against
Returns:
List of policy names that match the context
"""
matching: List[str] = []
from litellm.proxy.policy_engine.attachment_registry import (
get_attachment_registry,
)
for policy_name, policy in policies.items():
if PolicyMatcher.scope_matches(scope=policy.scope, context=context):
matching.append(policy_name)
verbose_proxy_logger.debug(
f"Policy '{policy_name}' matches context: "
f"team_alias={context.team_alias}, "
f"key_alias={context.key_alias}, "
f"model={context.model}"
)
registry = get_attachment_registry()
if not registry.is_initialized():
verbose_proxy_logger.debug(
"AttachmentRegistry not initialized, returning empty list"
)
return []
return matching
return registry.get_attached_policies(context)
@staticmethod
def get_matching_policies_from_registry(
@ -121,13 +122,4 @@ class PolicyMatcher:
Returns:
List of policy names that match the context
"""
from litellm.proxy.policy_engine.policy_registry import get_policy_registry
registry = get_policy_registry()
if not registry.is_initialized():
return []
return PolicyMatcher.get_matching_policies(
policies=registry.get_all_policies(),
context=context,
)
return PolicyMatcher.get_matching_policies(context=context)

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

@ -2,16 +2,21 @@
Policy Registry - In-memory storage for policies.
Handles storing, retrieving, and managing policies.
Policies define WHAT guardrails to apply. WHERE they apply is defined
by policy_attachments (see AttachmentRegistry).
"""
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,
PolicyScope,
PolicyStatement,
)
@ -21,6 +26,11 @@ class PolicyRegistry:
This is a singleton that holds all loaded policies and provides
methods to access them.
Policies define WHAT guardrails to apply:
- Base guardrails via guardrails.add/remove
- Inheritance via inherit field
- Conditional guardrails via statements
"""
def __init__(self):
@ -73,20 +83,103 @@ class PolicyRegistry:
# Handle legacy format where guardrails might be a list
guardrails = PolicyGuardrails(add=guardrails_data if guardrails_data else None)
# Parse scope
scope_data = policy_data.get("scope", {})
scope = PolicyScope(
teams=scope_data.get("teams"),
keys=scope_data.get("keys"),
models=scope_data.get("models"),
)
# 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
]
return Policy(
inherit=policy_data.get("inherit"),
guardrails=guardrails,
scope=scope,
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", []),
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.

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

@ -2,12 +2,13 @@
Policy Resolver - Resolves final guardrail list from policies.
Handles:
- Inheritance chain resolution
- Inheritance chain resolution (inherit with add/remove)
- Applying add/remove guardrails
- Evaluating conditional statements
- Combining guardrails from multiple matching policies
"""
from typing import Dict, List, Optional, Set
from typing import Any, Dict, List, Optional, Set
from litellm._logging import verbose_proxy_logger
from litellm.types.proxy.policy_engine import (
@ -21,7 +22,9 @@ class PolicyResolver:
"""
Resolves the final list of guardrails from policies.
Handles inheritance chains and add/remove operations.
Handles:
- Inheritance chains with add/remove operations
- Conditional statements with AWS IAM-style conditions
"""
@staticmethod
@ -68,13 +71,22 @@ class PolicyResolver:
def resolve_policy_guardrails(
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.
This method:
1. Resolves the inheritance chain
2. Applies add/remove from each policy in the chain
3. Evaluates conditional statements (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
Returns:
ResolvedPolicy with final guardrails list
@ -92,36 +104,88 @@ class PolicyResolver:
if policy is None:
continue
# Add guardrails
# Add guardrails from guardrails.add
for guardrail in policy.guardrails.get_add():
guardrails.add(guardrail)
# Remove guardrails
# Remove guardrails from guardrails.remove
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.
This:
1. Finds all policies that match the context
1. Finds all policies that match the context via policy_attachments
2. Resolves each policy's guardrails (including inheritance)
3. Combines all guardrails (union)
3. Evaluates conditional statements
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
@ -135,10 +199,8 @@ class PolicyResolver:
return []
policies = registry.get_all_policies()
# Get matching policies
matching_policy_names = PolicyMatcher.get_matching_policies(
policies=policies, context=context
)
# Get matching policies via attachments
matching_policy_names = PolicyMatcher.get_matching_policies(context=context)
if not matching_policy_names:
verbose_proxy_logger.debug(
@ -152,7 +214,10 @@ class PolicyResolver:
for policy_name in matching_policy_names:
resolved = PolicyResolver.resolve_policy_guardrails(
policy_name=policy_name, policies=policies
policy_name=policy_name,
policies=policies,
context=context,
metadata=metadata,
)
all_guardrails.update(resolved.guardrails)
verbose_proxy_logger.debug(
@ -169,6 +234,8 @@ class PolicyResolver:
@staticmethod
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.
@ -177,6 +244,8 @@ 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
Returns:
Dictionary mapping policy names to ResolvedPolicy objects
@ -193,7 +262,10 @@ class PolicyResolver:
for policy_name in policies:
resolved[policy_name] = PolicyResolver.resolve_policy_guardrails(
policy_name=policy_name, policies=policies
policy_name=policy_name,
policies=policies,
context=context,
metadata=metadata,
)
return resolved

4
litellm/proxy/proxy_server.py
View file

@ -2854,12 +2854,16 @@ class ProxyConfig:
llm_router: Optional LLM router for model validation
"""
if config is None:
verbose_proxy_logger.debug("Policy engine: config is None, skipping")
return
policies_config = config.get("policies", None)
if not policies_config:
verbose_proxy_logger.debug("Policy engine: no policies in config, skipping")
return
verbose_proxy_logger.info(f"Policy engine: found {len(policies_config)} policies in config")
from litellm.proxy.policy_engine.init_policies import init_policies
from litellm.proxy.policy_engine.policy_validator import PolicyValidator

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

@ -4,13 +4,24 @@ Type definitions for the LiteLLM Policy Engine.
The Policy Engine allows administrators to define policies that combine guardrails
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`.
"""
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,
@ -35,6 +46,11 @@ __all__ = [
"PolicyConfig",
"PolicyGuardrails",
"PolicyScope",
# Condition types (new)
"ConditionOperator",
"PolicyCondition",
"PolicyStatement",
"PolicyAttachment",
# Validation types
"PolicyValidateRequest",
"PolicyValidationError",

295
litellm/types/proxy/policy_engine/policy_types.py
View file

@ -2,16 +2,184 @@
Core policy type definitions.
These types define the structure of policies in the configuration.
Policy Engine Configuration:
```yaml
policies:
global-baseline:
description: "Base guardrails for all requests"
guardrails:
add: [pii_blocker]
healthcare-compliance:
inherit: global-baseline
guardrails:
add: [hipaa_audit]
statements:
- sid: "GPT4Only"
guardrails: [toxicity_filter]
condition:
model:
in: ["gpt-4", "gpt-4-turbo"]
policy_attachments:
- policy: global-baseline
scope: "*"
- policy: healthcare-compliance
teams: [healthcare-team]
```
Key concepts:
- `policies`: Define WHAT guardrails to apply (with inheritance via `inherit` and `guardrails.add`/`remove`)
- `policy_attachments`: Define WHERE policies apply (teams, keys, models)
- `statements`: Fine-grained conditional guardrails within a policy
"""
from typing import Dict, List, Optional
from typing import Any, Dict, List, Optional, Union
from pydantic import BaseModel, ConfigDict, Field
# ─────────────────────────────────────────────────────────────────────────────
# Condition Operators (AWS IAM-style)
# ─────────────────────────────────────────────────────────────────────────────
class ConditionOperator(BaseModel):
"""
AWS IAM-style condition operators for matching values.
Supports:
- equals: Exact string match
- in_: Value must be in the list (alias: "in" in YAML)
- prefix: Value must start with the given prefix
- not_equals: Value must NOT equal
- not_in: Value must NOT be in the list
Example YAML:
```yaml
condition:
model:
in: ["gpt-4", "gpt-4-turbo"]
team:
prefix: "healthcare-"
```
"""
equals: Optional[str] = Field(
default=None,
description="Exact string match.",
)
in_: Optional[List[str]] = Field(
default=None,
alias="in",
description="Value must be in this list.",
)
prefix: Optional[str] = Field(
default=None,
description="Value must start with this prefix.",
)
not_equals: Optional[str] = Field(
default=None,
description="Value must NOT equal this.",
)
not_in: Optional[List[str]] = Field(
default=None,
alias="notIn",
description="Value must NOT be in this list.",
)
model_config = ConfigDict(extra="forbid", populate_by_name=True)
class PolicyCondition(BaseModel):
"""
Condition for when a policy statement applies.
All specified conditions must match (AND logic).
If a field is None, it matches any value for that field.
Example YAML:
```yaml
condition:
model:
in: ["gpt-4", "gpt-4-turbo"]
team:
prefix: "healthcare-"
metadata:
environment:
equals: "production"
```
"""
model: Optional[ConditionOperator] = Field(
default=None,
description="Condition on the model name.",
)
team: Optional[ConditionOperator] = Field(
default=None,
description="Condition on the team alias.",
)
key: Optional[ConditionOperator] = Field(
default=None,
description="Condition on the API key alias.",
)
metadata: Optional[Dict[str, ConditionOperator]] = Field(
default=None,
description="Conditions on request metadata fields.",
)
model_config = ConfigDict(extra="forbid")
# ─────────────────────────────────────────────────────────────────────────────
# Policy Statements
# ─────────────────────────────────────────────────────────────────────────────
class PolicyStatement(BaseModel):
"""
A single statement within a policy.
Statements allow fine-grained control over when guardrails apply
using AWS IAM-style conditions.
Example YAML:
```yaml
statements:
- sid: "RequirePIIOnGPT4"
guardrails: [pii_blocker]
condition:
model:
in: ["gpt-4", "gpt-4-turbo"]
```
"""
sid: Optional[str] = Field(
default=None,
description="Statement ID for identification and debugging.",
)
guardrails: List[str] = Field(
default_factory=list,
description="Guardrail names to apply when condition matches.",
)
condition: Optional[PolicyCondition] = Field(
default=None,
description="Condition for when this statement applies. If None, always applies.",
)
model_config = ConfigDict(extra="forbid")
# ─────────────────────────────────────────────────────────────────────────────
# Policy Scope (used internally by attachments)
# ─────────────────────────────────────────────────────────────────────────────
class PolicyScope(BaseModel):
"""
Defines the scope for a policy - which requests it applies to.
Defines the scope for matching requests.
Used internally by PolicyAttachment to define WHERE a policy applies.
Scope Fields:
| Field | What it matches | Wildcard support |
@ -21,7 +189,7 @@ class PolicyScope(BaseModel):
| models | Model names | *, bedrock/*, gpt-* |
If a field is None or empty, it defaults to matching everything (["*"]).
A request must match ALL specified scope fields for the policy to apply.
A request must match ALL specified scope fields for the attachment to apply.
"""
teams: Optional[List[str]] = Field(
@ -52,12 +220,21 @@ class PolicyScope(BaseModel):
return self.models if self.models else ["*"]
# ─────────────────────────────────────────────────────────────────────────────
# Policy Guardrails
# ─────────────────────────────────────────────────────────────────────────────
class PolicyGuardrails(BaseModel):
"""
Defines guardrails to add or remove in a policy.
- `add`: List of guardrail names to add (on top of inherited guardrails)
- `remove`: List of guardrail names to remove (from inherited guardrails)
This supports the inheritance pattern where child policies can:
- Add new guardrails on top of parent's guardrails
- Remove specific guardrails inherited from parent
"""
add: Optional[List[str]] = Field(
@ -82,44 +259,63 @@ class PolicyGuardrails(BaseModel):
class Policy(BaseModel):
"""
A policy that defines which guardrails apply to requests matching its scope.
A policy that defines WHAT guardrails to apply.
Policies define guardrails but NOT where they apply - that's done via policy_attachments.
Policies can inherit from other policies using the `inherit` field.
When inheriting:
- Guardrails from `guardrails.add` are added to the inherited guardrails
- Guardrails from `guardrails.remove` are removed from the inherited guardrails
Policies can also have `statements` for fine-grained conditional guardrails.
Statements are evaluated in addition to the base guardrails.
Example configuration:
```yaml
policies:
global-baseline:
description: "Base guardrails for all requests"
guardrails:
add:
- pii_blocker
- phi_blocker
scope:
teams: ["*"]
keys: ["*"]
models: ["*"]
healthcare-compliance:
inherit: global-baseline
description: "HIPAA compliance for healthcare"
guardrails:
add:
- hipaa_audit
scope:
teams: [healthcare-team, medical-research]
models: [gpt-4, bedrock/claude-*]
internal-dev:
inherit: global-baseline
description: "Relaxed policy for dev"
guardrails:
add:
- toxicity_filter
remove:
- phi_blocker
scope:
keys: [dev-key-*, test-key-*]
conditional-policy:
description: "Model-specific guardrails"
guardrails:
add:
- base_guardrail
statements:
- sid: "GPT4Safety"
guardrails: [toxicity_filter]
condition:
model:
in: ["gpt-4", "gpt-4-turbo"]
policy_attachments:
- policy: global-baseline
scope: "*"
- policy: healthcare-compliance
teams: [healthcare-team]
- policy: internal-dev
keys: ["dev-key-*"]
```
"""
@ -131,14 +327,81 @@ class Policy(BaseModel):
default_factory=PolicyGuardrails,
description="Guardrails configuration with add/remove lists.",
)
scope: PolicyScope = Field(
default_factory=PolicyScope,
description="Scope defining which requests this policy applies to.",
statements: Optional[List[PolicyStatement]] = Field(
default=None,
description="Optional list of conditional statements for fine-grained guardrail control.",
)
description: Optional[str] = Field(
default=None,
description="Human-readable description of the policy.",
)
model_config = ConfigDict(extra="forbid")
# ─────────────────────────────────────────────────────────────────────────────
# Policy Attachments
# ─────────────────────────────────────────────────────────────────────────────
class PolicyAttachment(BaseModel):
"""
Attaches a policy to a scope - defines WHERE a policy applies.
Attachments are REQUIRED to make policies active. A policy without
an attachment will not be applied to any requests.
Example YAML:
```yaml
policy_attachments:
- policy: global-baseline
scope: "*" # applies to all requests
- policy: healthcare-compliance
teams: [healthcare-team, medical-research]
- policy: dev-safety
keys: ["dev-key-*", "test-key-*"]
- policy: gpt4-specific
models: ["gpt-4", "gpt-4-turbo"]
```
"""
policy: str = Field(
description="Name of the policy to attach.",
)
scope: Optional[str] = Field(
default=None,
description="Use '*' for global scope (applies to all requests).",
)
teams: Optional[List[str]] = Field(
default=None,
description="Team aliases or patterns this attachment applies to.",
)
keys: Optional[List[str]] = Field(
default=None,
description="Key aliases or patterns this attachment applies to.",
)
models: Optional[List[str]] = Field(
default=None,
description="Model names or patterns this attachment applies to.",
)
model_config = ConfigDict(extra="forbid")
def is_global(self) -> bool:
"""Check if this is a global attachment (scope='*')."""
return self.scope == "*"
def to_policy_scope(self) -> PolicyScope:
"""Convert attachment to a PolicyScope for matching."""
if self.is_global():
return PolicyScope(teams=["*"], keys=["*"], models=["*"])
return PolicyScope(
teams=self.teams,
keys=self.keys,
models=self.models,
)
class PolicyConfig(BaseModel):
"""
Root configuration for all policies.

95
proxy_config.yaml
View file

@ -2,6 +2,101 @@ model_list:
- model_name: "*"
litellm_params:
model: "*"
- model_name: "gpt-4"
litellm_params:
model: "gpt-4"
api_key: os.environ/OPENAI_API_KEY
- model_name: "gpt-3.5-turbo"
litellm_params:
model: "gpt-3.5-turbo"
api_key: os.environ/OPENAI_API_KEY
general_settings:
master_key: sk-1234
# ───────────────────────────────────────────────
# POLICIES - Define WHAT guardrails to apply
# ───────────────────────────────────────────────
#
# Policies define guardrails with:
# - inherit: Inherit guardrails from another policy
# - guardrails.add: Add guardrails (on top of inherited)
# - guardrails.remove: Remove guardrails (from inherited)
# - statements: Conditional guardrails with AWS IAM-style conditions
#
policies:
# Global baseline policy
global-baseline:
description: "Base guardrails for all requests"
guardrails:
add:
- pii_blocker
# Healthcare policy - inherits from global-baseline
healthcare-compliance:
inherit: global-baseline
description: "HIPAA compliance for healthcare teams"
guardrails:
add:
- hipaa_audit
# Dev policy - inherits but removes PII blocker for testing
internal-dev:
inherit: global-baseline
description: "Relaxed policy for internal development"
guardrails:
add:
- toxicity_filter
remove:
- pii_blocker
# Policy with conditional statements
conditional-safety:
description: "Model-specific guardrails using conditions"
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/"
# ───────────────────────────────────────────────
# POLICY ATTACHMENTS - Define WHERE policies apply
# ───────────────────────────────────────────────
#
# Attachments are REQUIRED to make policies active.
# A policy without an attachment will not be applied.
#
policy_attachments:
# Global attachment - applies to all requests
- policy: global-baseline
scope: "*"
# Team-specific attachment
- policy: healthcare-compliance
teams:
- healthcare-team
- medical-research
# Key pattern attachment
- policy: internal-dev
keys:
- "dev-key-*"
- "test-key-*"
# Model-specific attachment
- policy: conditional-safety
models:
- "gpt-4"
- "gpt-4-turbo"
- "bedrock/*"