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

1from __future__ import annotations 

2 

3import ipaddress 

4from collections.abc import Sequence 

5from typing import Any, Final 

6 

7from fastapi import Request 

8from pydantic import BaseModel, Field 

9 

10from litellm._logging import verbose_proxy_logger 

11 

12TrustedProxyNetwork = ipaddress.IPv4Network | ipaddress.IPv6Network 

13 

14 

15class NetworkContext(BaseModel): 

16 client_ip: str | None = None 

17 host: str | None = None 

18 via_trusted_proxy: bool = False 

19 

20 

21class TrustedProxyConfig(BaseModel): 

22 use_forwarded_for: bool = False 

23 trusted_proxy_cidrs: Sequence[str] = Field(default_factory=tuple) 

24 

25 

26def normalize_cidr_ranges(configured_ranges: Any, *, setting_name: str = "trusted_proxy_cidrs") -> list[str]: 

27 if not configured_ranges: 27 ↛ 29line 27 didn't jump to line 29 because the condition on line 27 was always true

28 return [] 

29 if isinstance(configured_ranges, str): 

30 return [r.strip() for r in configured_ranges.split(",") if r.strip()] 

31 if isinstance(configured_ranges, (list, tuple, set)): 

32 return [str(r).strip() for r in configured_ranges if str(r).strip()] 

33 verbose_proxy_logger.warning( 

34 "Invalid %s value: expected a list of CIDR ranges, got %s", 

35 setting_name, 

36 type(configured_ranges).__name__, 

37 ) 

38 return [] 

39 

40 

41def parse_trusted_proxy_ranges( 

42 configured_ranges: Any, *, setting_name: str = "trusted_proxy_cidrs" 

43) -> list[TrustedProxyNetwork]: 

44 networks: Final[list[TrustedProxyNetwork]] = [] 

45 for cidr in normalize_cidr_ranges(configured_ranges, setting_name=setting_name): 45 ↛ 46line 45 didn't jump to line 46 because the loop on line 45 never started

46 try: 

47 networks.append(ipaddress.ip_network(cidr, strict=False)) 

48 except ValueError: 

49 verbose_proxy_logger.warning("Invalid CIDR in %s: %s, skipping", setting_name, cidr) 

50 return networks 

51 

52 

53def _unmapped(addr: ipaddress.IPv4Address | ipaddress.IPv6Address) -> ipaddress.IPv4Address | ipaddress.IPv6Address: 

54 if isinstance(addr, ipaddress.IPv6Address) and addr.ipv4_mapped is not None: 

55 return addr.ipv4_mapped 

56 return addr 

57 

58 

59def ip_in_networks(client_ip: str | None, networks: list[TrustedProxyNetwork]) -> bool: 

60 if not client_ip or not networks: 

61 return False 

62 try: 

63 addr: Final = ipaddress.ip_address(client_ip.strip()) 

64 except ValueError: 

65 return False 

66 candidates: Final = (addr, _unmapped(addr)) 

67 return any(candidate in network for candidate in candidates for network in networks) 

68 

69 

70def _is_valid_ip(value: str) -> bool: 

71 try: 

72 ipaddress.ip_address(value) 

73 return True 

74 except ValueError: 

75 return False 

76 

77 

78def resolve_client_ip(request: Request, config: TrustedProxyConfig) -> tuple[str | None, bool]: 

79 """Resolve the real client IP, trusting X-Forwarded-For only when the direct 

80 peer is itself a configured trusted proxy. Walks the header right-to-left and 

81 returns the first hop that is not a trusted proxy, so a forged left-most entry 

82 cannot spoof the client.""" 

83 peer: Final = request.client.host if request.client else None 

84 networks: Final = parse_trusted_proxy_ranges(config.trusted_proxy_cidrs) 

85 if not config.use_forwarded_for or not ip_in_networks(peer, networks): 85 ↛ 87line 85 didn't jump to line 87 because the condition on line 85 was always true

86 return peer, False 

87 forwarded: Final = request.headers.get("x-forwarded-for", "") 

88 hops: Final = [h.strip() for h in forwarded.split(",") if h.strip()] 

89 for hop in reversed(hops): 

90 if _is_valid_ip(hop) and not ip_in_networks(hop, networks): 

91 return hop, True 

92 return peer, True 

93 

94 

95def resolve_network_context(request: Request, config: TrustedProxyConfig) -> NetworkContext: 

96 ip, via_proxy = resolve_client_ip(request, config) 

97 return NetworkContext( 

98 client_ip=ip, 

99 host=request.headers.get("host"), 

100 via_trusted_proxy=via_proxy, 

101 )