This commit is contained in:
Mateo Wang 2026-11-07 06:12:01 +00:00 • committed by GitHub
commit ef7bfe01ff
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 2252 additions and 158 deletions

View file

@ -149,7 +149,19 @@ def handle_anthropic_text_model_custom_llm_provider(
def declared_provider(model: str | None, custom_llm_provider: str | None = None) -> str | None:
return custom_llm_provider or (model.split("/", 1)[0] if model and "/" in model else None)
"""The provider this pair names outright, or None when only get_llm_provider can tell.
get_llm_provider reads a named provider's API key and base on the way to returning it, and a hosted
secret manager answers each read with a network call, so a caller that needs only the name reads it here.
"""
if custom_llm_provider:
return custom_llm_provider
prefix, slash, _ = (model or "").partition("/")
if not slash:
return None
if prefix in litellm.provider_list or JSONProviderRegistry.exists(prefix) or is_registered_custom_provider(prefix):
return prefix
return None
def declared_authenticating_provider(model: str | None, custom_llm_provider: str | None = None) -> str | None:

View file

@ -5413,6 +5413,79 @@ async def _granted_model_lists(
)
async def _user_object_or_none(
valid_token: UserAPIKeyAuth,
prisma_client: PrismaClient,
user_api_key_cache: UserApiKeyCache,
proxy_logging_obj: ProxyLogging,
) -> LiteLLM_UserTable | None:
try:
return await get_user_object(
user_id=valid_token.user_id,
prisma_client=prisma_client,
user_api_key_cache=user_api_key_cache,
user_id_upsert=False,
proxy_logging_obj=proxy_logging_obj,
)
except UserNotFoundError:
return None
async def enforced_model_allowlists(
valid_token: UserAPIKeyAuth,
prisma_client: PrismaClient | None,
user_api_key_cache: UserApiKeyCache,
proxy_logging_obj: ProxyLogging,
) -> tuple[Sequence[str], ...]:
"""One model allowlist per level that ``common_checks`` enforces on a request from this identity."""
key_models: Final = _resolve_key_models_for_auth_check(valid_token=valid_token)
if prisma_client is None:
return (key_models, tuple(valid_token.team_models or ()))
team_object: Final = (
None
if valid_token.team_id is None
else await get_team_object(
team_id=valid_token.team_id,
prisma_client=prisma_client,
user_api_key_cache=user_api_key_cache,
proxy_logging_obj=proxy_logging_obj,
)
)
user_object: Final = (
None
if team_object is not None
else await _user_object_or_none(
valid_token=valid_token,
prisma_client=prisma_client,
user_api_key_cache=user_api_key_cache,
proxy_logging_obj=proxy_logging_obj,
)
)
project_object: Final = (
None
if valid_token.project_id is None
else await get_project_object(
project_id=valid_token.project_id,
prisma_client=prisma_client,
user_api_key_cache=user_api_key_cache,
proxy_logging_obj=proxy_logging_obj,
)
)
return (
key_models,
team_object.models if team_object is not None else (),
await _team_member_granted_models(
valid_token=valid_token,
team_object=team_object,
prisma_client=prisma_client,
user_api_key_cache=user_api_key_cache,
proxy_logging_obj=proxy_logging_obj,
),
user_object.models if user_object is not None else (),
project_object.models if project_object is not None else (),
)
async def collect_matched_model_access_groups(
model: str | Sequence[str] | None,
valid_token: UserAPIKeyAuth | None,
@ -5726,6 +5799,45 @@ async def can_key_call_resolved_model(
)
async def can_caller_call_resolved_model(
model: str,
valid_token: UserAPIKeyAuth,
llm_router: litellm.Router | None,
prisma_client: PrismaClient | None,
user_api_key_cache: UserApiKeyCache,
proxy_logging_obj: ProxyLogging,
) -> None:
"""``can_key_call_resolved_model`` plus the agent grant and a personal key owner's own models, which
``common_checks`` applies to a model named at auth time and a model named after auth needs as well"""
await can_key_call_resolved_model(
model=model,
llm_model_list=None if llm_router is None else llm_router.registry().deployments(),
valid_token=valid_token,
llm_router=llm_router,
)
await can_agent_call_model(
model=model,
valid_token=valid_token,
llm_router=llm_router,
prisma_client=prisma_client,
user_api_key_cache=user_api_key_cache,
proxy_logging_obj=proxy_logging_obj,
)
if valid_token.team_id is not None or prisma_client is None:
return
await can_user_call_model(
model=model,
llm_router=llm_router,
user_object=await _user_object_or_none(
valid_token=valid_token,
prisma_client=prisma_client,
user_api_key_cache=user_api_key_cache,
proxy_logging_obj=proxy_logging_obj,
),
key_model_aliases=key_model_aliases_for_auth_check(valid_token),
)
def can_org_access_model(
model: str,
org_object: LiteLLM_OrganizationTable | None,

View file

@ -53,6 +53,7 @@ from litellm.integrations.custom_guardrail import guardrail_request_data_with_st
from litellm.litellm_core_utils.aws_partition import get_aws_dns_suffix
from litellm.litellm_core_utils.core_helpers import as_object_list, as_str_mapping, attribute_or_none
from litellm.litellm_core_utils.fallback_generalizations import match_routing_generalization
from litellm.litellm_core_utils.get_llm_provider_logic import declared_provider, inferred_provider
from litellm.llms.anthropic.common_utils import (
AnthropicModelInfo,
anthropic_pass_through_api_base,
@ -89,11 +90,16 @@ from litellm.passthrough.main import AsyncPassthroughStreamingResponse, is_provi
from litellm.proxy._types import *
from litellm.proxy.auth.auth_checks import (
can_agent_call_model,
can_caller_call_resolved_model,
can_key_call_resolved_model,
enforced_model_allowlists,
key_model_aliases_for_auth_check,
personal_key_team_model_access_applies,
team_model_aliases_for_auth_check,
)
from litellm.proxy.auth.auth_utils import log_model_access_denial
from litellm.proxy.auth.handle_jwt import JWTHandler
from litellm.proxy.auth.model_access_denied import model_access_denied_client_message
from litellm.proxy.auth.route_checks import RouteChecks
from litellm.proxy.auth.user_api_key_auth import ( # noqa: F401 # legacy module exports
_get_bearer_token, # pyright: ignore[reportPrivateUsage,reportUnusedImport] # backwards-compatible package export
@ -143,7 +149,10 @@ from litellm.proxy.pass_through_endpoints.llm_provider_handlers.openai_websocket
)
from litellm.proxy.pass_through_endpoints.pass_through_endpoints import (
STORED_PROMPT_PROBE_MODEL,
ClientFrameRefusal,
HttpPassThroughEndpointHelpers,
LoggedClientFrameRefusal,
WebsocketClientFrameGate,
WebsocketFrameModelAllowed,
WebsocketFrameModelGate,
WebsocketFrameModelRefused,
@ -189,6 +198,7 @@ from .passthrough_endpoint_router import PassthroughEndpointRouter
if TYPE_CHECKING:
from litellm.proxy.proxy_server import ProxyConfig as _ProxyConfig
from litellm.router import Router
from litellm.types.router import DeploymentTypedDict
ProxyConfig = _ProxyConfig # rebind-ok: conditional type alias
else:
@ -4107,6 +4117,12 @@ VERTEX_LIVE_UNCONFIGURED_CLOSE_REASON: Final = (
"Vertex AI auth failed: set a use_in_pass_through vertex model, default_vertex_config, or DEFAULT_VERTEXAI_* env"
)
VERTEX_LIVE_UNVERIFIED_MODEL_CLOSE_REASON: Final = (
"Could not check your key's access to the model this setup frame names; ask a proxy admin to read the logs"
)
VERTEX_LIVE_UNREADABLE_SETUP_CLOSE_REASON: Final = "Send setup frames as strict JSON so the proxy can check their model"
VERTEX_PUBLISHER_MODEL_PREFIX: Final = "publishers/google/models/"
VERTEX_PUBLISHERS_SEGMENT: Final = "publishers/"
@ -4131,6 +4147,361 @@ def _get_llm_router() -> Router | None:
return llm_router
@dataclass(frozen=True, slots=True)
class _UnreadableVertexLiveFrame:
pass
class _VertexLiveSetup(TypedDict, total=False):
model: ReadOnly[str]
class _VertexLiveClientFrame(TypedDict, total=False):
setup: ReadOnly[_VertexLiveSetup]
_VERTEX_LIVE_CLIENT_FRAME: Final = TypeAdapter(_VertexLiveClientFrame)
def _may_carry_vertex_live_setup(frame_data: str | bytes) -> bool:
"""
Whether a frame is worth parsing: audio and video frames stream continuously and are large, and one that
cannot spell ``setup`` cannot carry one. Vertex also reads single-quoted and bare keys, so the bare word is
the test, and a backslash means the key may be escaped, which only a parse settles
"""
if isinstance(frame_data, bytes):
return b"setup" in frame_data or b"\\" in frame_data
return "setup" in frame_data or "\\" in frame_data
def _object_with_unique_keys(pairs: Sequence[tuple[str, object]]) -> Mapping[str, object]:
unique: Final = MappingProxyType(dict(pairs))
if len(unique) < len(pairs):
raise ValueError("a JSON object repeats a key")
return unique
def _vertex_live_setup_model(frame_data: str | bytes) -> str | _UnreadableVertexLiveFrame | None:
"""
The model a Vertex AI Live ``setup`` frame names, which is the only place BidiGenerateContent carries one.
Vertex parses trailing commas, single quotes and bare keys that ``json.loads`` refuses, and keeps the first of
two keys when the second is null, so a frame that may carry a setup but fails a strict parse, repeats a key,
or holds a setup of the wrong shape is unreadable rather than setup-free
"""
if not _may_carry_vertex_live_setup(frame_data):
return None
try:
frame: Final = _VERTEX_LIVE_CLIENT_FRAME.validate_python(
json.loads(frame_data, object_pairs_hook=_object_with_unique_keys)
)
except (ValueError, RecursionError):
return _UnreadableVertexLiveFrame()
setup: Final = frame.get("setup")
return None if setup is None else setup.get("model") or None
def _deployment_named(model_name: str, deployments: Sequence[DeploymentTypedDict]) -> DeploymentTypedDict | None:
return next((deployment for deployment in deployments if deployment.get("model_name") == model_name), None)
VERTEX_LIVE_PROVIDERS: Final = frozenset({LlmProviders.VERTEX_AI.value, LlmProviders.VERTEX_AI_BETA.value})
def _upstream_model_and_provider(deployment: DeploymentTypedDict) -> tuple[str, str | None] | None:
litellm_params: Final = deployment["litellm_params"]
if "model" not in litellm_params:
return None
upstream: Final = litellm_params["model"]
declared: Final = declared_provider(upstream, litellm_params.get("custom_llm_provider"))
provider: Final = declared or inferred_provider(upstream)
return (upstream if provider is None else upstream.removeprefix(f"{provider}/"), provider)
def _vertex_upstream_model(deployment: DeploymentTypedDict) -> str | None:
"""
The publisher model a Vertex AI deployment runs, or ``None`` for any other provider's deployment, since a
group another provider serves grants nothing on the proxy's Vertex credentials
"""
resolved: Final = _upstream_model_and_provider(deployment)
if resolved is None:
return None
upstream, provider = resolved
return upstream if provider is None or provider in VERTEX_LIVE_PROVIDERS else None
def _routed_deployments(llm_router: Router) -> tuple[DeploymentTypedDict, ...]:
"""Every deployment under the name a request reaches it by, which leaves out a group a router alias shares a name
with, since the alias wins"""
return (
*llm_router.registry().deployments(source="alias"),
*(
deployment
for deployment in llm_router.registry().deployments() or ()
if deployment["model_name"] not in llm_router.registry().aliases()
),
)
def _vertex_deployments(llm_router: Router | None) -> tuple[DeploymentTypedDict, ...]:
if llm_router is None:
return ()
return tuple(deployment for deployment in _routed_deployments(llm_router) if _vertex_upstream_model(deployment))
def _live_deployment(setup_model: str, llm_router: Router | None) -> DeploymentTypedDict | None:
"""
The deployment whose group a ``setup`` frame's model names, whichever addressing the client used.
The Live SDK wraps whatever the caller typed as ``models/<name>``, and clients also send LiteLLM ids and
full Vertex resource paths, so the group is read from the last segment when the whole string names none
"""
if setup_model.startswith("projects/") or llm_router is None:
return None
deployments: Final = tuple(llm_router.registry().deployments() or ())
exact: Final = _deployment_named(setup_model, deployments)
return exact if exact is not None else _deployment_named(setup_model.rsplit("/", 1)[-1], deployments)
def _vertex_live_model_names(setup_model: str, llm_router: Router | None) -> tuple[str, ...]:
"""
The names a key can hold for the model a ``setup`` frame runs: the Vertex AI group the setup rewriter sends it
through, else every Vertex AI group serving the publisher model Vertex runs, and either way that model's id bare
or under LiteLLM's ``vertex_ai/`` prefix unless a group holds that name.
A full resource path skips the rewriter and reaches Vertex as sent, so only its publisher model counts, and
a wildcard is matched against that id alone so addressing in front of it cannot satisfy the pattern
"""
deployment: Final = _live_deployment(setup_model, llm_router)
if deployment is not None and (upstream := _vertex_upstream_model(deployment)) is not None:
return (deployment["model_name"], *_publisher_model_id_names(upstream, llm_router))
opened_model: Final = _resolve_alias_to_upstream_model(setup_model, llm_router)
publisher_model: Final = _vertex_publisher_model_suffix(opened_model)
serving_groups: Final = tuple(
dict.fromkeys(
candidate["model_name"]
for candidate in _vertex_deployments(llm_router)
if _vertex_publisher_model_suffix(_vertex_upstream_model(candidate) or "") == publisher_model
)
)
return (*serving_groups, *_publisher_model_id_names(opened_model, llm_router))
def _publisher_model_id_names(model: str, llm_router: Router | None) -> tuple[str, ...]:
group_names: Final = frozenset(
candidate["model_name"]
for candidate in (() if llm_router is None else llm_router.registry().deployments() or ())
)
publisher_model_id: Final = model.rsplit("/", 1)[-1]
return tuple(name for name in (publisher_model_id, f"vertex_ai/{publisher_model_id}") if name not in group_names)
def _is_listed_router_alias(name: str, llm_router: Router) -> bool:
return any(deployment["model_name"] == name for deployment in llm_router.registry().deployments(source="alias"))
def _is_alias_the_rewriter_ignores(name: str, valid_token: UserAPIKeyAuth, llm_router: Router | None) -> bool:
return (
name in (key_model_aliases_for_auth_check(valid_token) or ())
or name in (team_model_aliases_for_auth_check(valid_token) or ())
or name in litellm.model_alias_map
or (
llm_router is not None
and name in llm_router.registry().aliases()
and not _is_listed_router_alias(name, llm_router)
)
)
def _as_model_denial(refusal: ProxyException | HTTPException) -> ProxyException:
match refusal:
case ProxyException():
return refusal
case HTTPException():
return ProxyException(
message=str(refusal.detail),
type=ProxyErrorTypes.auth_error,
param="model",
code=refusal.status_code,
)
async def _model_access_denial(
model: str, valid_token: UserAPIKeyAuth, llm_router: Router | None
) -> ProxyException | None:
from litellm.proxy.proxy_server import prisma_client, proxy_logging_obj, user_api_key_cache
try:
await can_caller_call_resolved_model(
model=model,
valid_token=valid_token,
llm_router=llm_router,
prisma_client=prisma_client,
user_api_key_cache=user_api_key_cache,
proxy_logging_obj=proxy_logging_obj,
)
except (ProxyException, HTTPException) as refusal:
return _as_model_denial(refusal)
return None
VertexLiveModelAuthorizer: TypeAlias = Callable[[str], Awaitable[UserAPIKeyAuth]]
async def _model_reauthorization_denial(
model: str, authorize_model: VertexLiveModelAuthorizer
) -> LoggedClientFrameRefusal | None:
from litellm.proxy.spend_tracking.budget_reservation import release_or_invalidate_budget_reservation
try:
model_auth: Final = await authorize_model(model)
except (ProxyException, HTTPException) as refusal:
return LoggedClientFrameRefusal(_as_model_denial(refusal))
await release_or_invalidate_budget_reservation(budget_reservation=model_auth.budget_reservation)
return None
async def _first_model_denial(
model_names: tuple[str, ...],
valid_token: UserAPIKeyAuth,
llm_router: Router | None,
authorize_model: VertexLiveModelAuthorizer,
) -> ClientFrameRefusal | None:
"""
Every name runs the same deployment, so the first name the key may call is re-authorized and that answer
stands, budget and pricing checks included. Only a key that may call none of them gets the first access denial
"""
name: Final = model_names[0]
access_denial: Final = await _model_access_denial(name, valid_token, llm_router)
if access_denial is None:
return await _model_reauthorization_denial(name, authorize_model)
if len(model_names) == 1:
return access_denial
later_denial: Final = await _first_model_denial(model_names[1:], valid_token, llm_router, authorize_model)
return access_denial if isinstance(later_denial, ProxyException) else later_denial
async def _vertex_live_model_denial(
setup_model: str,
valid_token: UserAPIKeyAuth,
llm_router: Router | None,
authorize_model: VertexLiveModelAuthorizer,
) -> ClientFrameRefusal | None:
"""
The setup rewriter runs a listed router alias as its target group, the way a request naming it does, and sends
any other alias name to Vertex as written, so only a listed router alias grants anything here. A name the key may
call is then authorized the way a request naming it at connect time is
"""
model_names: Final = tuple(
name
for name in _vertex_live_model_names(setup_model, llm_router)
if not _is_alias_the_rewriter_ignores(name, valid_token, llm_router)
)
if not model_names:
return (
ModelAccessDeniedProxyException(
message=model_access_denied_client_message(model=setup_model),
internal_message=f"Vertex AI Live setup model {setup_model} has no name a model limit can grant",
type=ProxyErrorTypes.key_model_access_denied,
param="model",
code=403,
)
if await _caller_has_model_limits(valid_token)
else None
)
return await _first_model_denial(model_names, valid_token, llm_router, authorize_model)
_ALL_MODEL_ACCESS: Final = frozenset(
{
SpecialModelNames.all_proxy_models.value,
SpecialModelNames.all_team_models.value,
"*",
}
)
def _has_model_restrictions(model_allowlists: tuple[Sequence[str], ...]) -> bool:
return any(not _ALL_MODEL_ACCESS.issuperset(map(str, allowlist)) for allowlist in model_allowlists)
async def _caller_has_model_limits(valid_token: UserAPIKeyAuth) -> bool:
from litellm.proxy.proxy_server import prisma_client, proxy_logging_obj, user_api_key_cache
return (
valid_token.agent_id is not None
or valid_token.managed_agent_policy is not None
or valid_token.agent_caller is not None
or valid_token.end_user_id is not None
or personal_key_team_model_access_applies(valid_token)
or _has_model_restrictions(
await enforced_model_allowlists(
valid_token=valid_token,
prisma_client=prisma_client,
user_api_key_cache=user_api_key_cache,
proxy_logging_obj=proxy_logging_obj,
)
)
)
async def _vertex_live_setup_denial(
setup_model: str | _UnreadableVertexLiveFrame,
valid_token: UserAPIKeyAuth,
llm_router: Router | None,
authorize_model: VertexLiveModelAuthorizer,
) -> ClientFrameRefusal | None:
match setup_model:
case str():
denial: Final = await _vertex_live_model_denial(setup_model, valid_token, llm_router, authorize_model)
if isinstance(denial, ProxyException):
log_model_access_denial(denial)
return denial
case _UnreadableVertexLiveFrame():
return (
ProxyException(
message=VERTEX_LIVE_UNREADABLE_SETUP_CLOSE_REASON,
type=ProxyErrorTypes.bad_request_error,
param="setup",
code=400,
)
if await _caller_has_model_limits(valid_token)
else None
)
def _build_vertex_live_client_frame_gate(
llm_router: Router | None, authorize_model: VertexLiveModelAuthorizer
) -> WebsocketClientFrameGate:
"""
Apply the checks a model named at auth time gets to the model a client's ``setup`` frame names.
``?model=`` is optional on this route and the Live protocol carries the real model in the first client
frame, so connect-time auth has no model to check and every frame that names one has to be authorized
here instead.
A frame the proxy cannot read the way Vertex does only reaches Vertex for a caller who may call any model,
which is a caller with no model allowlist and no agent, end user or team grant standing in for one
"""
async def gate(frame_data: str | bytes, valid_token: UserAPIKeyAuth, /) -> ClientFrameRefusal | None:
model: Final = _vertex_live_setup_model(frame_data)
if model is None:
return None
try:
return await _vertex_live_setup_denial(model, valid_token, llm_router, authorize_model)
except Exception: # noqa: BLE001 # fail closed: a gate that cannot answer refuses, and says so in band
verbose_proxy_logger.exception("Vertex AI Live passthrough: model access check failed for %s", model)
return ProxyException(
message=VERTEX_LIVE_UNVERIFIED_MODEL_CLOSE_REASON,
type=ProxyErrorTypes.internal_server_error,
param="model",
code=500,
)
return gate
def _resolve_vertex_live_credentials(
vertex_project: str | None,
vertex_location: str | None,
@ -4189,24 +4560,9 @@ def _resolve_alias_to_upstream_model(setup_model: str, llm_router: Router | None
"""
The Live SDK wraps whatever the caller typed as ``models/<name>``, so a gateway alias arrives prefixed
"""
if llm_router is None:
return setup_model
candidates: Final = (setup_model, setup_model.rsplit("/", 1)[-1])
upstream: Final = next(
(
deployment["litellm_params"].get("model")
for deployment in (llm_router.registry().deployments() or ())
if deployment.get("model_name") in candidates
),
None,
)
if upstream is None:
return setup_model
try:
_, provider, _, _ = litellm.get_llm_provider(model=upstream)
except litellm.exceptions.BadRequestError:
return upstream
return upstream.removeprefix(f"{provider}/")
deployment: Final = _live_deployment(setup_model, llm_router)
resolved: Final = None if deployment is None else _upstream_model_and_provider(deployment)
return setup_model if resolved is None else resolved[0]
async def vertex_ai_live_websocket_passthrough(
@ -4215,6 +4571,7 @@ async def vertex_ai_live_websocket_passthrough(
vertex_project: str | None = None,
vertex_location: str | None = None,
user_api_key_dict: UserAPIKeyAuth | None = None,
authorize_model: VertexLiveModelAuthorizer | None = None,
):
"""
Vertex AI Live API WebSocket Pass-through Function
@ -4320,6 +4677,10 @@ async def vertex_ai_live_websocket_passthrough(
vertex_location=resolved_location,
llm_router=_get_llm_router(),
),
client_frame_gate=_build_vertex_live_client_frame_gate(
llm_router=_get_llm_router(),
authorize_model=authorize_model or partial(user_api_key_auth_accepted_websocket_for_model, websocket),
),
)

View file

@ -12,7 +12,7 @@ from datetime import datetime
from functools import reduce
from itertools import chain, count, groupby, takewhile
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, Protocol, TypeAlias, TypedDict, TypeVar, cast
from typing import TYPE_CHECKING, Any, Final, Protocol, TypeAlias, TypedDict, TypeVar, cast, overload
from urllib.parse import parse_qs, urlencode, urlparse
import httpx
@ -2554,28 +2554,52 @@ def create_websocket_passthrough_route(
return websocket_endpoint_func
def _rewrite_vertex_live_setup_model(text_data: str, setup_model_rewriter: Callable[[str], str] | None) -> str:
@overload
def _rewrite_vertex_live_setup_model(frame_data: str, setup_model_rewriter: Callable[[str], str] | None) -> str: ...
@overload
def _rewrite_vertex_live_setup_model(frame_data: bytes, setup_model_rewriter: Callable[[str], str] | None) -> bytes: ...
def _rewrite_vertex_live_setup_model(
frame_data: str | bytes, setup_model_rewriter: Callable[[str], str] | None
) -> str | bytes:
"""
Rewrite the model of a Vertex AI Live ``setup`` frame, leaving every other frame byte-identical
"""
if setup_model_rewriter is None:
return text_data
return frame_data
try:
message: Final[object] = _DECODED_JSON.validate_python(json.loads(text_data))
except json.JSONDecodeError:
return text_data
message: Final[object] = _DECODED_JSON.validate_python(json.loads(frame_data))
except (json.JSONDecodeError, UnicodeDecodeError, RecursionError):
return frame_data
if not isinstance(message, dict):
return text_data
return frame_data
setup: Final = message.get("setup")
if not isinstance(setup, dict):
return text_data
return frame_data
setup_model: Final = setup.get("model")
if not isinstance(setup_model, str):
return text_data
return frame_data
rewritten_model: Final = setup_model_rewriter(setup_model)
if rewritten_model == setup_model:
return text_data
return json.dumps({**message, "setup": {**setup, "model": rewritten_model}})
return frame_data
rewritten_frame: Final = json.dumps({**message, "setup": {**setup, "model": rewritten_model}})
return rewritten_frame if isinstance(frame_data, str) else rewritten_frame.encode("utf-8")
def _vertex_live_client_setup_model(
frame_data: str | bytes, setup_model_rewriter: Callable[[str], str] | None
) -> str | None:
try:
message: Final = _json_object_frame(frame_data)
except RecursionError:
return None
setup: Final = None if message is None else message.get("setup")
if not isinstance(setup, dict) or "model" not in setup:
return None
return _extract_model_from_vertex_ai_setup(_resolved_vertex_live_setup(setup, setup_model_rewriter))
def _resolved_vertex_live_setup(
@ -2662,6 +2686,34 @@ def _with_trace_context(headers: Mapping[str, str], parent_span: object) -> dict
return inject_trace_context(headers, parent_span=parent_span)
@dataclass(frozen=True, slots=True)
class LoggedClientFrameRefusal:
denial: ProxyException
ClientFrameRefusal: TypeAlias = ProxyException | LoggedClientFrameRefusal
def _client_frame_denial(refusal: ClientFrameRefusal) -> ProxyException:
match refusal:
case LoggedClientFrameRefusal(denial=denial):
return denial
case ProxyException():
return refusal
class WebsocketClientFrameGate(Protocol):
"""Authorize one client frame, returning the denial to report, one its check already reported, or ``None`` to
let it through.
A websocket passthrough authenticates before any frame arrives, so a protocol that names its model
inside a frame has no model to authorize at connect time. The frame, not the extracted model, is the
argument because where a model sits is part of each provider's wire protocol
"""
async def __call__(self, frame_data: str | bytes, valid_token: UserAPIKeyAuth, /) -> ClientFrameRefusal | None: ...
@dataclass(frozen=True, slots=True)
class WebsocketFrameModelAllowed:
access_groups: tuple[str, ...]
@ -2992,10 +3044,22 @@ async def _relay_client_frame(
upstream_ws: ClientConnection,
checked_frame: str | bytes,
forwarded_frame: str | bytes,
gate: WebsocketFrameModelGate | None,
frame_gate: WebsocketClientFrameGate | None,
model_gate: WebsocketFrameModelGate | None,
valid_token: UserAPIKeyAuth,
ledger: _WebsocketTurnLedger,
) -> _ForwardClientFrame | _RefuseClientFrame:
verdict: Final = await _client_frame_verdict(websocket, checked_frame, gate)
) -> _ForwardClientFrame | _RefuseClientFrame | ClientFrameRefusal:
"""Forward one client frame upstream through the gate its route set, or refuse it.
A frame gate's denial travels out so the caller reports a failed session instead of leaving the success hooks to
record a refusal as a success, and closing either socket is then the caller's job: awaiting a close here lets the
upstream reader finish first, which cancels this task mid-close and loses the denial. A model gate's refusal is
answered in band with the error event its protocol carries
"""
frame_refusal: Final = None if frame_gate is None else await frame_gate(checked_frame, valid_token)
if frame_refusal is not None:
return frame_refusal
verdict: Final = await _client_frame_verdict(websocket, checked_frame, model_gate)
match verdict:
case _RefuseClientFrame(refusal=refusal):
await refuse_websocket(websocket, refusal)
@ -3032,6 +3096,7 @@ async def websocket_passthrough_request(
cost_per_request: float | None = None,
accept_websocket: bool = True,
setup_model_rewriter: Callable[[str], str] | None = None,
client_frame_gate: WebsocketClientFrameGate | None = None,
client_frame_model_gate: WebsocketFrameModelGate | None = None,
upstream_frame_kind: WebsocketUpstreamFrameClassifier | None = None,
walkout_turn_drain_seconds: float = WALKOUT_TURN_DRAIN_SECONDS,
@ -3051,6 +3116,7 @@ async def websocket_passthrough_request(
endpoint: The endpoint path (for logging purposes)
cost_per_request: Optional field - cost per request to the target endpoint
setup_model_rewriter: Optional rewrite of the setup frame's model before it reaches the upstream
client_frame_gate: Optional per-frame authorization, applied before a frame reaches the upstream
client_frame_model_gate: Optional per-frame check of every model a client frame names
upstream_frame_kind: Optional sort of upstream frames into the one that ends a billable turn and the ones
that open the session, so each turn logs its own spend and keeps the session's provider-named model
@ -3158,6 +3224,21 @@ async def websocket_passthrough_request(
kwargs: Final = init_call_kwargs(user_api_key_dict, passthrough_logging_payload, litellm_call_id, logging_obj)
async def log_failure(exception: Exception, traceback_str: str | None = None) -> None:
await proxy_logging_obj.post_call_failure_hook(
user_api_key_dict=user_api_key_dict,
original_exception=exception,
request_data=_build_passthrough_failure_request_payload(
parsed_body=None if model is None else {"model": model},
kwargs=kwargs,
logging_obj=logging_obj,
custom_llm_provider=None,
url=httpx.URL(target),
builtin_provider_route=builtin_provider_route,
),
traceback_str=traceback_str,
)
def start_call_logging(call_logging: LiteLLMLoggingObj, call_kwargs: Mapping[str, object]) -> None:
call_logging.update_environment_variables(
model="unknown",
@ -3289,8 +3370,36 @@ async def websocket_passthrough_request(
"WebSocket passthrough (%s): Upstream connection established successfully", endpoint
)
async def forward_client_to_upstream() -> None:
"""Forward messages from client to upstream WebSocket"""
is_vertex_live: Final = bool(endpoint and "/vertex_ai/live" in endpoint)
def record_client_setup_model(frame_data: str | bytes) -> None:
if not is_vertex_live:
return
extracted_model: Final = _vertex_live_client_setup_model(frame_data, setup_model_rewriter)
if not extracted_model:
return
kwargs["model"] = extracted_model
kwargs["custom_llm_provider"] = "vertex_ai-language-models"
logging_obj.model = extracted_model
logging_obj.model_call_details["model"] = extracted_model
logging_obj.model_call_details["custom_llm_provider"] = "vertex_ai"
async def relay_client_frame(
frame_data: str | bytes,
) -> _ForwardClientFrame | _RefuseClientFrame | ClientFrameRefusal:
return await _relay_client_frame(
websocket,
upstream_ws,
frame_data,
_rewrite_vertex_live_setup_model(frame_data, setup_model_rewriter),
client_frame_gate,
client_frame_model_gate,
user_api_key_dict,
ledger,
)
async def forward_client_to_upstream() -> ClientFrameRefusal | None:
"""Forward messages from client to upstream WebSocket, returning any frame gate denial that ended it"""
nonlocal metered_total
try:
while True:
@ -3303,83 +3412,22 @@ async def websocket_passthrough_request(
bytes_data: bytes | None = message.get("bytes")
if text_data is not None:
# Try to extract model from client setup message for Vertex AI Live
if endpoint and "/vertex_ai/live" in endpoint:
verbose_proxy_logger.debug(
"WebSocket passthrough (%s): Processing client message for model extraction",
endpoint,
)
try:
client_message = _DECODED_JSON.validate_python(json.loads(text_data))
if isinstance(client_message, dict) and "setup" in client_message:
setup_data = client_message["setup"]
verbose_proxy_logger.debug(
"WebSocket passthrough (%s): Found setup data in client message: %s",
endpoint,
setup_data,
)
if isinstance(setup_data, dict) and "model" in setup_data:
extracted_model = _extract_model_from_vertex_ai_setup(
_resolved_vertex_live_setup(setup_data, setup_model_rewriter)
)
if extracted_model:
kwargs["model"] = extracted_model
kwargs["custom_llm_provider"] = "vertex_ai-language-models"
# Update logging object with correct model
logging_obj.model = extracted_model
logging_obj.model_call_details["model"] = extracted_model
logging_obj.model_call_details["custom_llm_provider"] = "vertex_ai"
verbose_proxy_logger.info(
"WebSocket passthrough (%s): Successfully extracted model '%s' and set provider to 'vertex_ai' from client setup message",
endpoint,
extracted_model,
)
else:
verbose_proxy_logger.warning(
"WebSocket passthrough (%s): Failed to extract model from client setup data: %s",
endpoint,
setup_data,
)
else:
verbose_proxy_logger.debug(
"WebSocket passthrough (%s): Setup data does not contain model field: %s",
endpoint,
setup_data,
)
else:
verbose_proxy_logger.debug(
"WebSocket passthrough (%s): Client message does not contain setup data",
endpoint,
)
except (json.JSONDecodeError, KeyError, TypeError) as e:
verbose_proxy_logger.debug(
"WebSocket passthrough (%s): Client message is not a valid setup message: %s",
endpoint,
e,
)
# Not a JSON message or doesn't contain setup data
if isinstance(
await _relay_client_frame(
websocket,
upstream_ws,
text_data,
_rewrite_vertex_live_setup_model(text_data, setup_model_rewriter),
client_frame_model_gate,
ledger,
),
_RefuseClientFrame,
):
break
metered_total += _metered_client_text(meter, text_data)
match await relay_client_frame(text_data):
case _RefuseClientFrame():
break
case _ForwardClientFrame():
record_client_setup_model(text_data)
metered_total += _metered_client_text(meter, text_data)
case text_refusal:
return text_refusal
elif bytes_data is not None:
if isinstance(
await _relay_client_frame(
websocket, upstream_ws, bytes_data, bytes_data, client_frame_model_gate, ledger
),
_RefuseClientFrame,
):
break
match await relay_client_frame(bytes_data):
case _RefuseClientFrame():
break
case _ForwardClientFrame():
record_client_setup_model(bytes_data)
case bytes_refusal:
return bytes_refusal
if not _turn_owed(upstream_frame_kind, websocket_messages, ledger):
await upstream_ws.close()
except asyncio.CancelledError:
@ -3405,11 +3453,13 @@ async def websocket_passthrough_request(
logging_obj.model_call_details["model"] = extracted_model
logging_obj.model_call_details["custom_llm_provider"] = "vertex_ai_language_models"
is_vertex_live: Final = bool(endpoint and "/vertex_ai/live" in endpoint)
json_frame_ordinal: Final = count()
def record_upstream_frame(upstream_message: str | bytes) -> None:
message_data: Final = _json_object_frame(upstream_message)
try:
message_data: Final = _json_object_frame(upstream_message)
except RecursionError:
return
if message_data is None:
return
if is_vertex_live and next(json_frame_ordinal) == 0:
@ -3495,13 +3545,33 @@ async def websocket_passthrough_request(
except asyncio.CancelledError:
pass
task_results: Final = tuple(task.result() for task in done if task.exception() is None)
# A refusal outranks any exception it caused: closing the client socket is what makes the
# upstream reader's next send fail, so raising that instead would report the wrong cause.
# Turns that completed before a refusal were served, so they are logged and billed as a success
client_frame_refusal: Final = next(
(result for result in task_results if isinstance(result, (ProxyException, LoggedClientFrameRefusal))),
None,
)
if client_frame_refusal is not None:
await websocket.close(
code=CloseCode.POLICY_VIOLATION,
reason=_refusal_close_reason(_client_frame_denial(client_frame_refusal).message),
)
await upstream_ws.close()
if isinstance(client_frame_refusal, ProxyException) and not websocket_messages:
await log_failure(client_frame_refusal)
if client_frame_refusal is not None and not websocket_messages:
return
# Check for exceptions in completed tasks
for task in done:
exception = task.exception()
if exception is not None:
if exception is not None and client_frame_refusal is None:
raise exception
upstream_close: Final = _upstream_close_to_relay(task.result() for task in done)
upstream_close: Final = _upstream_close_to_relay(task_results)
if upstream_close is not None and _client_socket_is_open(websocket):
await websocket.close(
code=upstream_close.code,
@ -3531,21 +3601,7 @@ async def websocket_passthrough_request(
except InvalidStatus as exc:
verbose_proxy_logger.exception("WebSocket passthrough (%s): upstream rejected WebSocket connection", endpoint)
await proxy_logging_obj.post_call_failure_hook(
user_api_key_dict=user_api_key_dict,
original_exception=exc,
request_data=_build_passthrough_failure_request_payload(
parsed_body=None if model is None else {"model": model},
kwargs=kwargs,
logging_obj=logging_obj,
custom_llm_provider=None,
url=httpx.URL(target),
builtin_provider_route=builtin_provider_route,
),
traceback_str=traceback.format_exc(
limit=MAXIMUM_TRACEBACK_LINES_TO_LOG,
),
)
await log_failure(exc, traceback.format_exc(limit=MAXIMUM_TRACEBACK_LINES_TO_LOG))
if _client_socket_is_open(websocket):
await websocket.close(
@ -3557,21 +3613,7 @@ async def websocket_passthrough_request(
"WebSocket passthrough (%s): unexpected error while proxying WebSocket", endpoint
)
await proxy_logging_obj.post_call_failure_hook(
user_api_key_dict=user_api_key_dict,
original_exception=e,
request_data=_build_passthrough_failure_request_payload(
parsed_body=None if model is None else {"model": model},
kwargs=kwargs,
logging_obj=logging_obj,
custom_llm_provider=None,
url=httpx.URL(target),
builtin_provider_route=builtin_provider_route,
),
traceback_str=traceback.format_exc(
limit=MAXIMUM_TRACEBACK_LINES_TO_LOG,
),
)
await log_failure(e, traceback.format_exc(limit=MAXIMUM_TRACEBACK_LINES_TO_LOG))
if _client_socket_is_open(websocket):
await websocket.close(code=1011, reason="WebSocket passthrough error")

View file

@ -11,6 +11,7 @@ import litellm
from litellm import CustomLLM
from litellm.litellm_core_utils import get_llm_provider_logic
from litellm.litellm_core_utils.get_llm_provider_logic import (
declared_provider,
get_llm_provider,
resolve_model_and_provider_for_metadata,
get_provider_affinity_header,
@ -103,6 +104,26 @@ def test_is_registered_custom_provider(registered_custom_provider: str, candidat
assert is_registered_custom_provider(candidate) is expected
@pytest.mark.parametrize(
("model", "custom_llm_provider", "expected"),
[
("vertex_ai/gemini-live-2.5-flash", None, "vertex_ai"),
("llama-4-scout", "groq", "groq"),
("gemini/gemini-2.0-flash-live-001", "vertex_ai", "vertex_ai"),
("veniceai/venice-uncensored", None, "veniceai"),
(f"{CUSTOM_PROVIDER}/my-model", None, CUSTOM_PROVIDER),
("gemini-live-2.5-flash", None, None),
("models/gemini-live-2.5-flash", None, None),
],
ids=["provider-prefix", "custom-llm-provider", "custom-llm-provider-wins", "json-provider", "custom-provider",
"bare-name", "unknown-prefix"],
)
def test_declared_provider_names_only_a_provider_the_pair_spells_out(
registered_custom_provider: str, model: str, custom_llm_provider: str | None, expected: str | None
) -> None:
assert declared_provider(model, custom_llm_provider) == expected
def test_get_llm_provider_leaves_fal_ai_api_base_unset_for_global_fallback() -> None:
_, provider, _, api_base = get_llm_provider(model="fal_ai/fal-ai/flux/schnell")
assert provider == "fal_ai"

View file

@ -1,7 +1,7 @@
# What is this?
## Tests if 'get_end_user_object' works as expected
import sys, os, asyncio, time, random, uuid
import sys, os, asyncio, time, random, re, uuid
import hashlib
import json
import traceback
@ -41,6 +41,7 @@ from litellm.proxy.utils import PrismaClient
import litellm.proxy.common_utils.auth_cache_invalidation_pubsub as pubsub_module
from litellm.caching.in_memory_cache import InMemoryCache
from litellm.proxy.auth.auth_checks import (
can_caller_call_resolved_model,
can_team_access_model,
delete_cache_key_objects,
is_model_cost_zero,
@ -2423,6 +2424,37 @@ async def test_bulk_eviction_clears_the_redis_copies_before_waiting_on_a_wedged_
assert redis.redis == {}
@pytest.mark.asyncio
@pytest.mark.parametrize(("team_id", "refused"), [(None, True), ("live-team", False)], ids=["personal-key", "team-key"])
async def test_can_caller_call_resolved_model_holds_a_personal_key_to_its_owners_models(
team_id: str | None, refused: bool
) -> None:
owner_id: Final = f"live-owner-{uuid.uuid4()}"
cache: Final = UserApiKeyCache()
await cache.async_set_cache(key=owner_id, value=LiteLLM_UserTable(user_id=owner_id, models=["gemini-3.8-live"]))
router: Final = litellm.Router(
model_list=[
{"model_name": "gemini-3.8-live", "litellm_params": {"model": "vertex_ai/gemini-3.8-live"}},
{"model_name": "gemini-live-2.5-flash", "litellm_params": {"model": "vertex_ai/gemini-live-2.5-flash"}},
]
)
call: Final = can_caller_call_resolved_model(
model="gemini-live-2.5-flash",
valid_token=UserAPIKeyAuth(api_key="sk-live", user_id=owner_id, team_id=team_id),
llm_router=router,
prisma_client=MagicMock(),
user_api_key_cache=cache,
proxy_logging_obj=MagicMock(),
)
if refused:
with pytest.raises(ProxyException, match=re.escape("gemini-live-2.5-flash")):
await call
else:
await call
class _SignallingPublishRedisClient(_SlowPublishRedisClient):
def __init__(self) -> None:
super().__init__()

View file

@ -11023,6 +11023,139 @@ async def test_key_budget_error_keeps_the_masked_key_name(key_name):
assert f"Key=prod-key ({key_name}) Current cost" in message
class _UntouchedPrisma:
def __getattr__(self, name: str) -> object:
raise AssertionError(f"database reached through {name}")
class _MissingUserPrisma:
class db:
class litellm_usertable:
@staticmethod
async def find_unique(where: dict[str, str], include: dict[str, bool]) -> None:
return None
@pytest.mark.asyncio
async def test_enforced_model_allowlists_treats_a_missing_user_row_as_unrestricted():
from litellm.proxy.auth.auth_checks import enforced_model_allowlists
from litellm.proxy.common_utils.user_api_key_cache import UserApiKeyCache
from litellm.proxy.utils import ProxyLogging
cache = UserApiKeyCache()
scopes = await enforced_model_allowlists(
valid_token=UserAPIKeyAuth(token="hashed-fake", user_id="default_user_id"),
prisma_client=_MissingUserPrisma(),
user_api_key_cache=cache,
proxy_logging_obj=ProxyLogging(user_api_key_cache=cache),
)
assert [list(scope) for scope in scopes] == [[], [], [], [], []]
class _UnreachableUserPrisma:
class db:
class litellm_usertable:
@staticmethod
async def find_unique(where: dict[str, str], include: dict[str, bool]) -> None:
raise RuntimeError("database gone")
@pytest.mark.asyncio
async def test_enforced_model_allowlists_surfaces_a_failed_user_lookup():
from litellm.proxy.auth.auth_checks import enforced_model_allowlists
from litellm.proxy.common_utils.user_api_key_cache import UserApiKeyCache
from litellm.proxy.utils import ProxyLogging
cache = UserApiKeyCache()
with pytest.raises(ValueError, match="database gone"):
await enforced_model_allowlists(
valid_token=UserAPIKeyAuth(token="hashed-fake", user_id="user-fake"),
prisma_client=_UnreachableUserPrisma(),
user_api_key_cache=cache,
proxy_logging_obj=ProxyLogging(user_api_key_cache=cache),
)
@pytest.mark.asyncio
async def test_enforced_model_allowlists_reads_every_level_from_cache():
from litellm.proxy._types import (
LiteLLM_BudgetTable,
LiteLLM_ProjectTableCachedObj,
LiteLLM_TeamMembership,
LiteLLM_TeamTableCachedObj,
LiteLLM_UserTable,
)
from litellm.proxy.auth.auth_checks import enforced_model_allowlists
from litellm.proxy.common_utils.user_api_key_cache import (
UserApiKeyCache,
team_membership_reservation_cache_key,
)
from litellm.proxy.utils import ProxyLogging
cache = UserApiKeyCache()
proxy_logging_obj = ProxyLogging(user_api_key_cache=cache)
await cache.async_set_cache(
key="team_id:team-fake", value=LiteLLM_TeamTableCachedObj(team_id="team-fake", models=["gpt-4o"])
)
await cache.async_set_cache(
key=team_membership_reservation_cache_key(user_id="user-fake", team_id="team-fake"),
value=LiteLLM_TeamMembership(
user_id="user-fake",
team_id="team-fake",
litellm_budget_table=LiteLLM_BudgetTable(allowed_models=["gpt-4o-mini"]),
),
)
await cache.async_set_cache(
key="project_id:project-fake",
value=LiteLLM_ProjectTableCachedObj(project_id="project-fake", models=["gpt-4.1"]),
)
await cache.async_set_cache(key="user-fake", value=LiteLLM_UserTable(user_id="user-fake", models=["o3"]))
prisma_client = _UntouchedPrisma()
team_scoped = await enforced_model_allowlists(
valid_token=UserAPIKeyAuth(
token="hashed-fake",
models=["all-team-models"],
team_models=["gpt-4o", "gpt-4o-mini"],
user_id="user-fake",
team_id="team-fake",
project_id="project-fake",
),
prisma_client=prisma_client,
user_api_key_cache=cache,
proxy_logging_obj=proxy_logging_obj,
)
personal = await enforced_model_allowlists(
valid_token=UserAPIKeyAuth(token="hashed-fake", user_id="user-fake"),
prisma_client=prisma_client,
user_api_key_cache=cache,
proxy_logging_obj=proxy_logging_obj,
)
without_database = await enforced_model_allowlists(
valid_token=UserAPIKeyAuth(
token="hashed-fake",
models=["gpt-4o"],
team_models=["gpt-4o-mini"],
user_id="user-fake",
team_id="team-fake",
),
prisma_client=None,
user_api_key_cache=cache,
proxy_logging_obj=proxy_logging_obj,
)
assert [list(scope) for scope in team_scoped] == [
["gpt-4o", "gpt-4o-mini"],
["gpt-4o"],
["gpt-4o-mini"],
[],
["gpt-4.1"],
]
assert [list(scope) for scope in personal] == [[], [], [], ["o3"], []]
assert [list(scope) for scope in without_database] == [["gpt-4o"], ["gpt-4o-mini"]]
@pytest.mark.asyncio
@pytest.mark.parametrize("channel", ["team", "key"])
async def test_access_group_model_fallback_uses_the_injected_database(channel: str) -> None: