Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/auth/fallback_model_access.py: 46%

1""" 

2Authorize router fallback targets against the caller's key, team and project model access. 

3 

4`_enforce_key_and_fallback_model_access` only sees fallbacks the client sends in the request body. 

5Fallbacks configured on the router (`router_settings.fallbacks` and friends) are chosen after auth, 

6inside the router, so this predicate is injected into the router to re-run the same model access 

7checks for each fallback target before it is attempted. Opt-in via 

8`general_settings.enforce_fallback_model_access: true`. 

9""" 

10 

11from collections.abc import Callable, Mapping 

12from dataclasses import dataclass 

13from typing import Final 

14 

15from pydantic import BaseModel, ValidationError 

16 

17from litellm._logging import verbose_proxy_logger 

18from litellm.proxy._types import ProxyException, UserAPIKeyAuth 

19from litellm.proxy.auth.auth_checks import can_key_call_resolved_model 

20from litellm.router import Router 

21 

22 

23class _RequestMetadata(BaseModel): 

24 user_api_key_auth: UserAPIKeyAuth | None = None 

25 

26 

27class _FallbackAccessSettings(BaseModel): 

28 enforce_fallback_model_access: bool = False 

29 

30 

31async def is_model_authorized_for_token(*, model: str, valid_token: UserAPIKeyAuth, llm_router: Router) -> bool: 

32 try: 

33 await can_key_call_resolved_model( 

34 model=model, 

35 llm_model_list=None, 

36 valid_token=valid_token, 

37 llm_router=llm_router, 

38 ) 

39 except ProxyException: 

40 return False 

41 except Exception as e: # noqa: BLE001 # fail closed: a lookup failure must neither run the fallback nor replace the provider error 

42 verbose_proxy_logger.warning("Skipping fallback to model=%s: authorization lookup failed: %s", model, e) 

43 return False 

44 return True 

45 

46 

47def _token_in_metadata(metadata: object) -> UserAPIKeyAuth | None: 

48 try: 

49 return _RequestMetadata.model_validate(metadata).user_api_key_auth 

50 except ValidationError: 

51 return None 

52 

53 

54def _user_api_key_auth_from_request(request_kwargs: Mapping[str, object]) -> UserAPIKeyAuth | None: 

55 return next( 

56 ( 

57 token 

58 for field in ("metadata", "litellm_metadata") 

59 if (token := _token_in_metadata(request_kwargs.get(field))) is not None 

60 ), 

61 None, 

62 ) 

63 

64 

65def _enforced_by_general_settings() -> bool: 

66 from litellm.proxy.proxy_server import general_settings 

67 

68 return _FallbackAccessSettings.model_validate(general_settings).enforce_fallback_model_access 

69 

70 

71@dataclass(frozen=True, slots=True) 

72class RouterFallbackAccessCheck: 

73 """ 

74 `FallbackAccessCheck` for the proxy's router: while `is_enforced()` is true, a fallback target 

75 is attempted only when the key behind the request could have requested it directly. Requests 

76 that carry no key (for example internal health checks) are not restricted. 

77 """ 

78 

79 is_enforced: Callable[[], bool] 

80 

81 async def __call__(self, *, model: str, request_kwargs: Mapping[str, object], llm_router: Router) -> bool: 

82 if not self.is_enforced(): 

83 return True 

84 valid_token: Final = _user_api_key_auth_from_request(request_kwargs) 

85 if valid_token is None: 

86 return True 

87 return await is_model_authorized_for_token(model=model, valid_token=valid_token, llm_router=llm_router) 

88 

89 

90router_fallback_access_check: Final = RouterFallbackAccessCheck(is_enforced=_enforced_by_general_settings)