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

1from typing import Any, Final 

2 

3from fastapi import Request 

4 

5from litellm._logging import verbose_proxy_logger 

6from litellm.proxy.auth.network import ( 

7 ip_in_networks, 

8 normalize_cidr_ranges, 

9 parse_trusted_proxy_ranges, 

10) 

11 

12TRUSTED_PROXY_RANGES_KEY: Final = "trusted_proxy_ranges" 

13 

14 

15def _get_proxy_general_settings() -> dict[str, Any]: 

16 try: 

17 from litellm.proxy.proxy_server import general_settings 

18 

19 return general_settings or {} 

20 except ImportError: 

21 return {} 

22 

23 

24def get_trusted_proxy_cidrs( 

25 general_settings: dict[str, Any] | None = None, 

26) -> list[str]: 

27 """Operator-configured trusted reverse-proxy CIDRs, normalized to strings. 

28 

29 Empty when none are configured, in which case X-Forwarded-For must not be 

30 trusted and only the direct peer is authoritative. 

31 """ 

32 if general_settings is None: 32 ↛ 34line 32 didn't jump to line 34 because the condition on line 32 was always true

33 general_settings = _get_proxy_general_settings() 

34 return normalize_cidr_ranges( 

35 general_settings.get(TRUSTED_PROXY_RANGES_KEY), 

36 setting_name=TRUSTED_PROXY_RANGES_KEY, 

37 ) 

38 

39 

40def _get_direct_client_ip(request: Request) -> str | None: 

41 client: Final = getattr(request, "client", None) 

42 client_host: Final = getattr(client, "host", None) 

43 if isinstance(client_host, str): 

44 return client_host 

45 return None 

46 

47 

48def require_trusted_proxy_request( 

49 *, 

50 request: Request, 

51 general_settings: dict[str, Any] | None = None, 

52 feature_name: str, 

53 setting_name: str = TRUSTED_PROXY_RANGES_KEY, 

54) -> None: 

55 """ 

56 Fail closed unless the direct TCP peer is one of the configured 

57 trusted reverse proxies. 

58 

59 Header-based auth paths must validate the direct peer, not 

60 X-Forwarded-For, because the direct peer is the actor supplying the 

61 identity headers. 

62 """ 

63 if general_settings is None: 

64 general_settings = _get_proxy_general_settings() 

65 

66 trusted_networks: Final = parse_trusted_proxy_ranges(general_settings.get(setting_name), setting_name=setting_name) 

67 if not trusted_networks: 

68 raise ValueError( 

69 f"{feature_name} requires general_settings.{setting_name} before " 

70 "trusting identity headers from an upstream proxy." 

71 ) 

72 

73 direct_client_ip: Final = _get_direct_client_ip(request) 

74 if not ip_in_networks(direct_client_ip, trusted_networks): 

75 verbose_proxy_logger.warning( 

76 "%s rejected identity headers from untrusted direct client IP %r", 

77 feature_name, 

78 direct_client_ip, 

79 ) 

80 raise ValueError( 

81 f"{feature_name} only accepts identity headers from configured " 

82 f"trusted proxy ranges. Direct client IP {direct_client_ip!r} " 

83 "is not trusted." 

84 )