Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/policy_engine/policy_matcher.py: 86%
64 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1"""
2Policy Matcher - Matches requests against policy attachments.
4Uses existing wildcard pattern matching helpers to determine which policies
5apply to a given request based on team alias, key alias, and model.
7Policies are matched via policy_attachments which define WHERE each policy applies.
8"""
10from collections.abc import Callable, Sequence
11from typing import Final
13from litellm._logging import verbose_proxy_logger
14from litellm.proxy.auth.route_checks import RouteChecks
15from litellm.proxy.policy_engine.policy_resolver import PolicyResolver
16from litellm.types.proxy.policy_engine import Policy, PolicyMatchContext, PolicyScope
19class PolicyMatcher:
20 """
21 Matches incoming requests against policy attachments.
23 Supports wildcard patterns:
24 - "*" matches everything
25 - "prefix-*" matches anything starting with "prefix-"
27 Uses policy_attachments to determine which policies apply to a request.
28 """
30 @staticmethod
31 def matches_pattern(value: str | None, patterns: list[str]) -> bool:
32 """
33 Check if a value matches any of the given patterns.
35 Uses the existing RouteChecks.route_matches_wildcard_pattern helper.
37 Args:
38 value: The value to check (e.g., team alias, key alias, model)
39 patterns: List of patterns to match against
41 Returns:
42 True if value matches any pattern, False otherwise
43 """
44 # If no value provided, only match if patterns include "*"
45 if value is None:
46 return "*" in patterns
48 for pattern in patterns:
49 # Use existing wildcard pattern matching helper
50 if RouteChecks.route_matches_wildcard_pattern(route=value, pattern=pattern):
51 return True
53 return False
55 @staticmethod
56 def scope_matches(scope: PolicyScope, context: PolicyMatchContext) -> bool:
57 """
58 Check if a policy scope matches the given context.
60 A scope matches if ALL of its fields match:
61 - teams matches context.team_alias
62 - keys matches context.key_alias
63 - models matches context.model
65 Args:
66 scope: The policy scope to check
67 context: The request context
69 Returns:
70 True if scope matches context, False otherwise
71 """
72 # Check teams
73 if not PolicyMatcher.matches_pattern(context.team_alias, scope.get_teams()): 73 ↛ 74line 73 didn't jump to line 74 because the condition on line 73 was never true
74 return False
76 # Check keys
77 if not PolicyMatcher.matches_pattern(context.key_alias, scope.get_keys()): 77 ↛ 78line 77 didn't jump to line 78 because the condition on line 77 was never true
78 return False
80 # Check models
81 if not PolicyMatcher.matches_pattern(context.model, scope.get_models()): 81 ↛ 82line 81 didn't jump to line 82 because the condition on line 81 was never true
82 return False
84 # Check tags (only if scope specifies tags)
85 # Unlike teams/keys/models, empty tags means "do not check" rather than "match all"
86 scope_tags: Final = scope.get_tags()
87 if scope_tags:
88 if not context.tags:
89 return False
90 # Match if ANY context tag matches ANY scope tag pattern
91 if not any(PolicyMatcher.matches_pattern(tag, scope_tags) for tag in context.tags):
92 return False
94 return True
96 @staticmethod
97 def get_matching_policies(
98 context: PolicyMatchContext,
99 ) -> list[str]:
100 """
101 Get list of policy names that match the given context via attachments.
103 Args:
104 context: The request context to match against
106 Returns:
107 List of policy names that match the context
108 """
109 from litellm.proxy.policy_engine.attachment_registry import (
110 get_attachment_registry,
111 )
113 registry: Final = get_attachment_registry()
114 if not registry.is_initialized(): 114 ↛ 115line 114 didn't jump to line 115 because the condition on line 114 was never true
115 verbose_proxy_logger.debug("AttachmentRegistry not initialized, returning empty list")
116 return []
118 return registry.get_attached_policies(context, PolicyMatcher.policy_applies(context))
120 @staticmethod
121 def get_matching_policies_from_registry(
122 context: PolicyMatchContext,
123 ) -> list[str]:
124 """
125 Get list of policy names that match the given context from the global registry.
127 Args:
128 context: The request context to match against
130 Returns:
131 List of policy names that match the context
132 """
133 return PolicyMatcher.get_matching_policies(context=context)
135 @staticmethod
136 def policy_applies(
137 context: PolicyMatchContext,
138 policies: dict[str, Policy] | None = None,
139 ) -> Callable[[str], bool]:
140 """
141 Predicate telling whether a policy exists and any policy in its
142 inheritance chain applies to the context. Admissions where the
143 policy's own condition missed but an ancestor applies are logged at
144 INFO, once per attachment scan.
145 """
146 resolved: Final = policies if policies is not None else PolicyMatcher._registry_policies()
148 def applies(policy_name: str) -> bool:
149 applying: Final = PolicyMatcher._applying_chain_members(
150 policy_name=policy_name, context=context, policies=resolved
151 )
152 if applying and policy_name not in applying: 152 ↛ 153line 152 didn't jump to line 153 because the condition on line 152 was never true
153 verbose_proxy_logger.info(
154 "Policy '%s' applied through ancestor '%s' although its own condition did not match "
155 "(team_alias=%s, key_alias=%s, model=%s)",
156 policy_name,
157 applying[0],
158 context.team_alias,
159 context.key_alias,
160 context.model,
161 )
162 return bool(applying)
164 return applies
166 @staticmethod
167 def _applying_chain_members(
168 policy_name: str,
169 context: PolicyMatchContext,
170 policies: dict[str, Policy],
171 ) -> tuple[str, ...]:
172 from litellm.proxy.policy_engine.condition_evaluator import ConditionEvaluator
174 chain: Final = PolicyResolver.resolve_inheritance_chain(policy_name=policy_name, policies=policies)
175 return tuple(
176 name
177 for name in chain
178 if (policy := policies.get(name)) is not None
179 and (policy.condition is None or ConditionEvaluator.evaluate(policy.condition, context))
180 )
182 @staticmethod
183 def _registry_policies() -> dict[str, Policy]:
184 from litellm.proxy.policy_engine.policy_registry import get_policy_registry
186 registry: Final = get_policy_registry()
187 return registry.get_all_policies() if registry.is_initialized() else {}
189 @staticmethod
190 def get_policies_with_matching_conditions(
191 policy_names: Sequence[str],
192 context: PolicyMatchContext,
193 policies: dict[str, Policy] | None = None,
194 ) -> list[str]:
195 """
196 Filter policies to only those that apply to the given context.
198 A policy applies when any policy in its inheritance chain has no
199 condition or a condition that evaluates to True for the context. The
200 resolver then drops only the chain members whose own condition fails,
201 so a child whose condition misses still contributes the guardrails of
202 its unconditional ancestors. A missing policy resolves to an empty
203 chain and does not apply.
205 Args:
206 policy_names: List of policy names to filter
207 context: The request context to evaluate conditions against
208 policies: Dictionary of all policies (if None, uses global registry)
210 Returns:
211 List of policy names that apply to the context
212 """
213 resolved: Final = policies if policies is not None else PolicyMatcher._registry_policies()
214 return [
215 policy_name
216 for policy_name in policy_names
217 if PolicyMatcher._applying_chain_members(policy_name, context, resolved)
218 ]