"""
Policy Validator - Validates policy configurations.
Validates:
- Guardrail names exist in the guardrail registry
- Non-wildcard team aliases exist in the database
- Non-wildcard key aliases exist in the database
- Non-wildcard model names exist in the router or match a wildcard route
- Inheritance chains are valid (no cycles, parents exist)
"""
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Set
from litellm._logging import verbose_proxy_logger
from litellm.types.proxy.policy_engine import (
Policy,
PolicyValidationError,
PolicyValidationErrorType,
PolicyValidationResponse,
)
if TYPE_CHECKING:
from litellm.proxy.utils import PrismaClient
from litellm.router import Router
class PolicyValidator:
"""
Validates policy configurations against actual data.
"""
def __init__(
self,
prisma_client: Optional["PrismaClient"] = None,
llm_router: Optional["Router"] = None,
):
"""
Initialize the validator.
Args:
prisma_client: Optional Prisma client for database validation
llm_router: Optional LLM router for model validation
"""
self.prisma_client = prisma_client
self.llm_router = llm_router
@staticmethod
def is_wildcard_pattern(pattern: str) -> bool:
"""
Check if a pattern contains wildcards.
Args:
pattern: The pattern to check
Returns:
True if the pattern contains wildcard characters
"""
return "*" in pattern or "?" in pattern
def get_available_guardrails(self) -> Set[str]:
"""
Get set of available guardrail names from the guardrail registry.
Returns:
Set of guardrail names
"""
try:
from litellm.proxy.guardrails.guardrail_registry import (
IN_MEMORY_GUARDRAIL_HANDLER,
)
guardrails = IN_MEMORY_GUARDRAIL_HANDLER.list_in_memory_guardrails()
return {g.get("guardrail_name", "") for g in guardrails if g.get("guardrail_name")}
except Exception as e:
verbose_proxy_logger.warning(
f"Could not get guardrails from registry: {str(e)}"
)
return set()
async def check_team_alias_exists(self, team_alias: str) -> bool:
"""
Check if a specific team alias exists in the database.
Args:
team_alias: The team alias to check
Returns:
True if the team alias exists
"""
if self.prisma_client is None:
return True # Can't validate without DB, assume valid
try:
team = await self.prisma_client.db.litellm_teamtable.find_first(
where={"team_alias": team_alias},
)
return team is not None
except Exception as e:
verbose_proxy_logger.warning(
f"Could not check team alias '{team_alias}': {str(e)}"
)
return True # Assume valid on error
async def check_key_alias_exists(self, key_alias: str) -> bool:
"""
Check if a specific key alias exists in the database.
Args:
key_alias: The key alias to check
Returns:
True if the key alias exists
"""
if self.prisma_client is None:
return True # Can't validate without DB, assume valid
try:
key = await self.prisma_client.db.litellm_verificationtoken.find_first(
where={"key_alias": key_alias},
)
return key is not None
except Exception as e:
verbose_proxy_logger.warning(
f"Could not check key alias '{key_alias}': {str(e)}"
)
return True # Assume valid on error
def check_model_exists(self, model: str) -> bool:
"""
Check if a model exists in the router or matches a wildcard pattern.
Args:
model: The model name to check
Returns:
True if the model exists or matches a pattern in the router
"""
if self.llm_router is None:
return True # Can't validate without router, assume valid
try:
# Check if model is in router's model names
if model in self.llm_router.model_names:
return True
# Check if model matches any pattern via pattern router
if hasattr(self.llm_router, "pattern_router"):
pattern_deployments = self.llm_router.pattern_router.get_deployments_by_pattern(
model=model
)
if pattern_deployments:
return True
return False
except Exception as e:
verbose_proxy_logger.warning(
f"Could not check model '{model}': {str(e)}"
)
return True # Assume valid on error
def _validate_inheritance_chain(
self,
policy_name: str,
policies: Dict[str, Policy],
visited: Optional[Set[str]] = None,
max_depth: int = 100,
) -> List[PolicyValidationError]:
"""
Validate the inheritance chain for a policy.
Checks for:
- Parent policy exists
- No circular inheritance
- Max depth not exceeded
Args:
policy_name: Name of the policy to validate
policies: All policies
visited: Set of already visited policy names (for cycle detection)
max_depth: Maximum recursion depth to prevent infinite loops
Returns:
List of validation errors
"""
errors: List[PolicyValidationError] = []
# Prevent infinite recursion
if max_depth <= 0:
errors.append(
PolicyValidationError(
policy_name=policy_name,
error_type=PolicyValidationErrorType.CIRCULAR_INHERITANCE,
message="Inheritance chain too deep (exceeded max depth of 100)",
field="inherit",
)
)
return errors
if visited is None:
visited = set()
if policy_name in visited:
errors.append(
PolicyValidationError(
policy_name=policy_name,
error_type=PolicyValidationErrorType.CIRCULAR_INHERITANCE,
message=f"Circular inheritance detected: {' -> '.join(visited)} -> {policy_name}",
field="inherit",
)
)
return errors
policy = policies.get(policy_name)
if policy is None:
return errors
if policy.inherit:
if policy.inherit not in policies:
errors.append(
PolicyValidationError(
policy_name=policy_name,
error_type=PolicyValidationErrorType.INVALID_INHERITANCE,
message=f"Parent policy '{policy.inherit}' not found",
field="inherit",
value=policy.inherit,
)
)
else:
# Recursively check parent with decremented depth
visited.add(policy_name)
errors.extend(
self._validate_inheritance_chain(
policy.inherit, policies, visited, max_depth - 1
)
)
return errors
async def validate_policies(
self,
policies: Dict[str, Policy],
validate_db: bool = True,
) -> PolicyValidationResponse:
"""
Validate a set of policies.
Args:
policies: Dictionary mapping policy names to Policy objects
validate_db: Whether to validate against database (teams, keys)
Returns:
PolicyValidationResponse with errors and warnings
"""
errors: List[PolicyValidationError] = []
warnings: List[PolicyValidationError] = []
# Get available guardrails
available_guardrails = self.get_available_guardrails()
for policy_name, policy in policies.items():
# Validate guardrails
for guardrail in policy.guardrails.get_add():
if available_guardrails and guardrail not in available_guardrails:
errors.append(
PolicyValidationError(
policy_name=policy_name,
error_type=PolicyValidationErrorType.INVALID_GUARDRAIL,
message=f"Guardrail '{guardrail}' not found in guardrail registry",
field="guardrails.add",
value=guardrail,
)
)
for guardrail in policy.guardrails.get_remove():
if available_guardrails and guardrail not in available_guardrails:
warnings.append(
PolicyValidationError(
policy_name=policy_name,
error_type=PolicyValidationErrorType.INVALID_GUARDRAIL,
message=f"Guardrail '{guardrail}' in remove list not found in guardrail registry",
field="guardrails.remove",
value=guardrail,
)
)
# Validate pipeline if present
if policy.pipeline is not None:
pipeline_errors = PolicyValidator._validate_pipeline(
policy_name=policy_name,
policy=policy,
available_guardrails=available_guardrails,
)
errors.extend(pipeline_errors)
# Validate inheritance
inheritance_errors = self._validate_inheritance_chain(
policy_name=policy_name, policies=policies
)
errors.extend(inheritance_errors)
return PolicyValidationResponse(
valid=len(errors) == 0,
errors=errors,
warnings=warnings,
)
@staticmethod
def _validate_pipeline(
policy_name: str,
policy: Policy,
available_guardrails: Set[str],
) -> List[PolicyValidationError]:
"""Validate a policy's pipeline configuration."""
errors: List[PolicyValidationError] = []
pipeline = policy.pipeline
if pipeline is None:
return errors
guardrails_add = set(policy.guardrails.get_add())
for i, step in enumerate(pipeline.steps):
# Check guardrail is in policy's guardrails.add
if step.guardrail not in guardrails_add:
errors.append(
PolicyValidationError(
policy_name=policy_name,
error_type=PolicyValidationErrorType.INVALID_GUARDRAIL,
message=(
f"Pipeline step {i} guardrail '{step.guardrail}' "
f"is not in the policy's guardrails.add list"
),
field="pipeline.steps",
value=step.guardrail,
)
)
# Check guardrail exists in registry
if available_guardrails and step.guardrail not in available_guardrails:
errors.append(
PolicyValidationError(
policy_name=policy_name,
error_type=PolicyValidationErrorType.INVALID_GUARDRAIL,
message=(
f"Pipeline step {i} guardrail '{step.guardrail}' "
f"not found in guardrail registry"
),
field="pipeline.steps",
value=step.guardrail,
)
)
return errors
async def validate_policy_config(
self,
policy_config: Dict[str, Any],
validate_db: bool = True,
) -> PolicyValidationResponse:
"""
Validate a raw policy configuration dictionary.
This parses the config and then validates it.
Args:
policy_config: Raw policy configuration from YAML
validate_db: Whether to validate against database
Returns:
PolicyValidationResponse with errors and warnings
"""
from litellm.proxy.policy_engine.policy_registry import PolicyRegistry
# First, try to parse the policies
errors: List[PolicyValidationError] = []
policies: Dict[str, Policy] = {}
temp_registry = PolicyRegistry()
for policy_name, policy_data in policy_config.items():
try:
policy = temp_registry._parse_policy(policy_name, policy_data)
policies[policy_name] = policy
except Exception as e:
errors.append(
PolicyValidationError(
policy_name=policy_name,
error_type=PolicyValidationErrorType.INVALID_SYNTAX,
message=f"Failed to parse policy: {str(e)}",
)
)
# If there were parsing errors, return early
if errors:
return PolicyValidationResponse(
valid=False,
errors=errors,
warnings=[],
)
# Validate the parsed policies
return await self.validate_policies(policies, validate_db=validate_db)