mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-19 00:01:29 +00:00
fix(tag_regex): address greptile review - security docs, strict-mode enforcement, validation order
- Strengthen security note in tag_routing.md: explicitly state User-Agent is client-supplied and can be set to any value; frame tag_regex as a traffic classification hint, not an access-control mechanism - Move tag_regex startup validation before _add_deployment() so an invalid pattern never leaves partial router state - Enforce match_any=False strict-tag policy: when a deployment has both tags and tag_regex and the strict tag check fails, skip the regex fallback rather than silently bypassing the operator's intent - Extract per-deployment match logic into _match_deployment() helper to keep get_deployments_for_tag() readable - Add two new tests: strict-mode blocks regex fallback, regex-only deployment still matches under match_any=False
This commit is contained in:
parent
1b36694749
commit
c009f4f18f
4 changed files with 145 additions and 53 deletions
|
|
@ -300,12 +300,13 @@ When a regex matches, `tag_routing` is written into request metadata and flows t
|
|||
|
||||
:::caution
|
||||
|
||||
`User-Agent` is set by the client and **can be spoofed**. `tag_regex` is designed for **routing convenience** — directing traffic from a known tool to the right backend — not for security isolation or access control.
|
||||
**`User-Agent` is a client-supplied header and can be set to any value.** Any API consumer can send `User-Agent: claude-code/1.0` regardless of whether they are actually using Claude Code.
|
||||
|
||||
If you need to restrict which users or teams can reach a deployment, use [API key / team scoping](./users) rather than (or in addition to) regex routing.
|
||||
Do not rely on `tag_regex` routing to enforce access controls or spend limits — use [team/key-based routing](./users) for that. `tag_regex` is a **traffic classification hint** (useful for billing visibility, capacity planning, and routing convenience), not a security boundary.
|
||||
|
||||
:::
|
||||
|
||||
|
||||
---
|
||||
|
||||
## ✨ Team based tag routing (Enterprise)
|
||||
|
|
|
|||
|
|
@ -6527,10 +6527,8 @@ class Router:
|
|||
)
|
||||
return None
|
||||
|
||||
deployment = self._add_deployment(deployment=deployment)
|
||||
|
||||
# Validate tag_regex patterns early so a bad regex fails at startup
|
||||
# rather than silently misbehaving on the first matching request.
|
||||
# Validate tag_regex patterns BEFORE adding the deployment so we never
|
||||
# have partially-initialised router state if a pattern is invalid.
|
||||
_tag_regex = deployment.litellm_params.get("tag_regex") or []
|
||||
for pattern in _tag_regex:
|
||||
try:
|
||||
|
|
@ -6541,6 +6539,8 @@ class Router:
|
|||
f"{pattern!r} — {exc}"
|
||||
) from exc
|
||||
|
||||
deployment = self._add_deployment(deployment=deployment)
|
||||
|
||||
model = deployment.to_json(exclude_none=True)
|
||||
|
||||
self._add_model_to_list_and_index_map(
|
||||
|
|
|
|||
|
|
@ -73,6 +73,54 @@ def is_valid_deployment_tag(
|
|||
return False
|
||||
|
||||
|
||||
def _match_deployment(
|
||||
deployment: Any,
|
||||
request_tags: Optional[List[str]],
|
||||
header_strings: List[str],
|
||||
match_any: bool,
|
||||
) -> Optional[Dict[str, str]]:
|
||||
"""
|
||||
Determine whether *deployment* matches the current request.
|
||||
|
||||
Returns {"matched_via": ..., "matched_value": ...} if the deployment
|
||||
should be included, or None if it should be excluded.
|
||||
|
||||
Priority:
|
||||
1. Exact tag match (respects match_any semantics).
|
||||
2. Regex match — skipped when match_any=False and the tag check already
|
||||
ran and failed, so the regex cannot override strict-tag policy.
|
||||
"""
|
||||
litellm_params = deployment.get("litellm_params", {})
|
||||
deployment_tags: Optional[List[str]] = litellm_params.get("tags")
|
||||
deployment_tag_regex: Optional[List[str]] = litellm_params.get("tag_regex")
|
||||
|
||||
# 1. Exact tag match (existing behaviour).
|
||||
if deployment_tags and request_tags:
|
||||
if is_valid_deployment_tag(deployment_tags, request_tags, match_any):
|
||||
matched_value = next(
|
||||
(t for t in deployment_tags if t in set(request_tags)),
|
||||
deployment_tags[0],
|
||||
)
|
||||
return {"matched_via": "tags", "matched_value": matched_value}
|
||||
|
||||
# 2. Regex match against request headers.
|
||||
# When match_any=False and the deployment has both plain tags and tag_regex,
|
||||
# the strict tag check has already failed (step 1 returned None). Allow
|
||||
# the regex to fire only when the deployment has NO plain tags, so we never
|
||||
# use regex as a backdoor around the operator's strict-tag policy.
|
||||
strict_tag_check_failed = (
|
||||
not match_any
|
||||
and bool(deployment_tags)
|
||||
and bool(request_tags)
|
||||
)
|
||||
if deployment_tag_regex and header_strings and not strict_tag_check_failed:
|
||||
regex_match = _is_valid_deployment_tag_regex(deployment_tag_regex, header_strings)
|
||||
if regex_match is not None:
|
||||
return {"matched_via": "tag_regex", "matched_value": regex_match}
|
||||
|
||||
return None
|
||||
|
||||
|
||||
async def get_deployments_for_tag(
|
||||
llm_router_instance: LitellmRouter,
|
||||
model: str, # used to raise the correct error
|
||||
|
|
@ -138,57 +186,21 @@ async def get_deployments_for_tag(
|
|||
user_agent,
|
||||
)
|
||||
for deployment in healthy_deployments:
|
||||
deployment_litellm_params = deployment.get("litellm_params")
|
||||
deployment_tags = deployment_litellm_params.get("tags")
|
||||
deployment_tag_regex = deployment_litellm_params.get("tag_regex")
|
||||
deployment_tags = deployment.get("litellm_params", {}).get("tags")
|
||||
|
||||
verbose_logger.debug(
|
||||
"deployment: %s tags: %s tag_regex: %s",
|
||||
deployment.get("model_name"),
|
||||
deployment_tags,
|
||||
deployment_tag_regex,
|
||||
match_result = _match_deployment(
|
||||
deployment=deployment,
|
||||
request_tags=request_tags,
|
||||
header_strings=header_strings,
|
||||
match_any=match_any,
|
||||
)
|
||||
|
||||
matched_via: Optional[str] = None
|
||||
matched_value: Optional[str] = None
|
||||
|
||||
# 1. Exact tag match (existing behaviour)
|
||||
if deployment_tags and request_tags:
|
||||
if is_valid_deployment_tag(deployment_tags, request_tags, match_any):
|
||||
matched_via = "tags"
|
||||
matched_value = next(
|
||||
(t for t in deployment_tags if t in set(request_tags)),
|
||||
deployment_tags[0],
|
||||
)
|
||||
|
||||
# 2. Regex match against request headers (new)
|
||||
# NOTE: tag_regex always uses OR semantics (any pattern match suffices).
|
||||
# match_any=False applies only to exact tag matching above; it has no
|
||||
# "all patterns must match" equivalent for regex and is intentionally
|
||||
# ignored here. Operators who need strict isolation should enforce
|
||||
# that via API key / team scoping rather than routing patterns alone,
|
||||
# since User-Agent is fully client-controlled and can be spoofed.
|
||||
if matched_via is None and deployment_tag_regex and header_strings:
|
||||
regex_match = _is_valid_deployment_tag_regex(
|
||||
deployment_tag_regex, header_strings
|
||||
)
|
||||
if regex_match is not None:
|
||||
if not match_any:
|
||||
verbose_logger.debug(
|
||||
"tag_regex match fired on deployment=%s while "
|
||||
"tag_filtering_match_any=False; regex routing always "
|
||||
"uses OR semantics — match_any is ignored for tag_regex",
|
||||
deployment.get("model_name"),
|
||||
)
|
||||
matched_via = "tag_regex"
|
||||
matched_value = regex_match
|
||||
|
||||
if matched_via is not None:
|
||||
if match_result is not None:
|
||||
verbose_logger.debug(
|
||||
"tag routing match: deployment=%s matched_via=%s matched_value=%s",
|
||||
deployment.get("model_name"),
|
||||
matched_via,
|
||||
matched_value,
|
||||
match_result["matched_via"],
|
||||
match_result["matched_value"],
|
||||
)
|
||||
# Record provenance in metadata so it flows to SpendLogs.
|
||||
# Written only for the first match — load balancer selects one
|
||||
|
|
@ -197,8 +209,8 @@ async def get_deployments_for_tag(
|
|||
if "tag_routing" not in metadata:
|
||||
metadata["tag_routing"] = {
|
||||
"matched_deployment": deployment.get("model_name"),
|
||||
"matched_via": matched_via,
|
||||
"matched_value": matched_value,
|
||||
"matched_via": match_result["matched_via"],
|
||||
"matched_value": match_result["matched_value"],
|
||||
"request_tags": request_tags or [],
|
||||
"user_agent": user_agent,
|
||||
}
|
||||
|
|
|
|||
|
|
@ -294,3 +294,82 @@ async def test_tag_routing_metadata_not_overwritten_for_multiple_matches():
|
|||
tr = metadata.get("tag_routing", {})
|
||||
assert tr.get("matched_deployment") == "claude-sonnet"
|
||||
assert tr.get("matched_via") == "tag_regex"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_match_any_false_strict_tag_check_blocks_regex_fallback():
|
||||
"""
|
||||
When match_any=False and a deployment has both tags and tag_regex:
|
||||
if the strict tag check fails (request has a tag NOT present on the
|
||||
deployment, so req_set is NOT a subset of dep_set), the regex fallback
|
||||
must NOT fire — that would violate the operator's strict-filtering intent.
|
||||
|
||||
Semantics of match_any=False: req_set.issubset(dep_set), i.e. every
|
||||
request tag must appear on the deployment. A request with tags ["vip"]
|
||||
against a deployment with tags ["premium"] fails because "vip" ∉ dep_set.
|
||||
"""
|
||||
deployment_strict = {
|
||||
"model_name": "claude-sonnet",
|
||||
"litellm_params": {
|
||||
"model": "openai/strict-deployment",
|
||||
"api_key": "fake",
|
||||
"tags": ["premium"],
|
||||
"tag_regex": [r"^User-Agent: claude-code\/"],
|
||||
},
|
||||
"model_info": {"id": "strict-deployment"},
|
||||
}
|
||||
default_deployment = {
|
||||
"model_name": "claude-sonnet",
|
||||
"litellm_params": {
|
||||
"model": "openai/default-deployment",
|
||||
"api_key": "fake",
|
||||
"tags": ["default"],
|
||||
},
|
||||
"model_info": {"id": "default-deployment"},
|
||||
}
|
||||
# match_any=False: req_set must be a subset of dep_set.
|
||||
# Request has "vip" which is NOT in ["premium"], so tag check fails.
|
||||
# Even though UA matches tag_regex, the deployment must NOT be selected.
|
||||
router = _make_router_mock(enable_tag_filtering=True, match_any=False)
|
||||
metadata: dict = {
|
||||
"tags": ["vip"], # "vip" not in deployment tags → strict check fails
|
||||
"user_agent": "claude-code/1.0",
|
||||
}
|
||||
result = await get_deployments_for_tag(
|
||||
llm_router_instance=router,
|
||||
model="claude-sonnet",
|
||||
healthy_deployments=[deployment_strict, default_deployment],
|
||||
request_kwargs={"metadata": metadata},
|
||||
)
|
||||
ids = [d["model_info"]["id"] for d in result]
|
||||
assert "strict-deployment" not in ids, (
|
||||
"strict-deployment should not be selected: strict tag check failed "
|
||||
"and regex must not override the strict policy"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_match_any_false_regex_only_deployment_still_matches():
|
||||
"""
|
||||
When match_any=False and a deployment has ONLY tag_regex (no plain tags),
|
||||
there is no strict tag policy to violate, so the regex check must still fire.
|
||||
"""
|
||||
regex_only_deployment = {
|
||||
"model_name": "claude-sonnet",
|
||||
"litellm_params": {
|
||||
"model": "openai/regex-only-deployment",
|
||||
"api_key": "fake",
|
||||
"tag_regex": [r"^User-Agent: claude-code\/"],
|
||||
# no "tags" key at all
|
||||
},
|
||||
"model_info": {"id": "regex-only-deployment"},
|
||||
}
|
||||
router = _make_router_mock(enable_tag_filtering=True, match_any=False)
|
||||
result = await get_deployments_for_tag(
|
||||
llm_router_instance=router,
|
||||
model="claude-sonnet",
|
||||
healthy_deployments=[regex_only_deployment],
|
||||
request_kwargs={"metadata": {"user_agent": "claude-code/1.0"}},
|
||||
)
|
||||
assert len(result) == 1
|
||||
assert result[0]["model_info"]["id"] == "regex-only-deployment"
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue