Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/middleware/per_request_root_path_middleware.py: 32%

1"""Per-request ``root_path`` resolution from ``SERVER_ROOT_PATHS``. 

2 

3``SERVER_ROOT_PATH`` is a startup scalar, so one deployment serves exactly one 

4client-visible URL path prefix; a request under any other prefix 404s before a 

5handler runs. When the ingress preserves several prefixes into one pod (e.g. 

6``/tenant-a/*`` and ``/tenant-b/*``), the matched prefix becomes that 

7request's ``scope["root_path"]`` instead: Starlette strips it during route 

8matching and rebuilds it into ``request.base_url``, so every emitted URL — 

9the MCP OAuth discovery ``resource`` (RFC 9728 §3) and the 401 challenges' 

10``resource_metadata`` among them — lands under the prefix the client called. 

11Opt-in: with ``SERVER_ROOT_PATHS`` unset the middleware is not added at all. 

12""" 

13 

14import os 

15from collections.abc import Sequence 

16from contextvars import ContextVar 

17from typing import Final 

18 

19from starlette.types import ASGIApp, Receive, Scope, Send 

20 

21from litellm._logging import verbose_proxy_logger 

22 

23SERVER_ROOT_PATHS_ENV: Final = "SERVER_ROOT_PATHS" 

24 

25# The effective ``root_path`` for the currently-handled request. Populated by 

26# ``PerRequestRootPathMiddleware`` from the (possibly-mutated) scope so code 

27# that emits URLs off the request path — the 401 challenges' resource_metadata 

28# and ``get_custom_url``'s SSO callbacks among them — can pick up the prefix 

29# the client actually called without threading scope through every call site. 

30# ``None`` means "middleware did not run" (the ``SERVER_ROOT_PATHS`` env is 

31# unset, so no per-request prefix exists); readers fall back to the scalar 

32# ``SERVER_ROOT_PATH`` in that case, which matches the pre-middleware behavior. 

33_request_root_path_var: Final[ContextVar[str | None]] = ContextVar("_request_root_path_var", default=None) 

34 

35 

36def get_request_root_path() -> str: 

37 """Return the effective ``root_path`` for the current request. 

38 

39 Reads the value ``PerRequestRootPathMiddleware`` stashed for this request; 

40 falls back through :func:`~litellm.proxy.utils.get_server_root_path` (i.e. 

41 the ``SERVER_ROOT_PATH`` env) when the middleware did not run — the 

42 scalar-only deployment. Delegating to the existing helper keeps every 

43 existing ``monkeypatch.setattr("litellm.proxy.utils.get_server_root_path"`` 

44 test override working, and keeps a single source of truth for the scalar. 

45 """ 

46 value: Final = _request_root_path_var.get() 

47 if value is not None: 47 ↛ 48line 47 didn't jump to line 48 because the condition on line 47 was never true

48 return value 

49 # Lazy import: utils.py imports this module (via the lazy import inside 

50 # get_custom_url), so a top-level import would build a cycle at load time. 

51 from litellm.proxy.utils import get_server_root_path # noqa: PLC0415 # lazy import breaks a two-way dep 

52 

53 return get_server_root_path() 

54 

55 

56def normalize_root_paths(raw_paths: Sequence[str]) -> tuple[str, ...]: 

57 """Strip whitespace and trailing slashes, dedupe, order longest-first; 

58 warn and drop entries missing a leading ``/`` and the bare root.""" 

59 kept: Final[list[str]] = [] # mutable-ok: local accumulator; escapes only as a tuple 

60 for entry in raw_paths: 

61 candidate = entry.strip() 

62 if not candidate: 

63 continue 

64 if not candidate.startswith("/"): 

65 verbose_proxy_logger.warning( 

66 "%s entry %r does not start with '/' and will be ignored.", 

67 SERVER_ROOT_PATHS_ENV, 

68 entry, 

69 ) 

70 continue 

71 candidate = candidate.rstrip("/") 

72 if not candidate: 

73 verbose_proxy_logger.warning( 

74 "%s entry %r is the bare root and will be ignored; a root-mounted deployment needs no entry.", 

75 SERVER_ROOT_PATHS_ENV, 

76 entry, 

77 ) 

78 continue 

79 if candidate not in kept: 

80 kept.append(candidate) 

81 return tuple(sorted(kept, key=len, reverse=True)) 

82 

83 

84def get_server_root_paths() -> tuple[str, ...]: 

85 """The normalized ``SERVER_ROOT_PATHS`` prefixes, empty when unset.""" 

86 configured: Final = os.getenv(SERVER_ROOT_PATHS_ENV, "") 

87 if not configured.strip(): 87 ↛ 89line 87 didn't jump to line 89 because the condition on line 87 was always true

88 return () 

89 return normalize_root_paths(configured.split(",")) 

90 

91 

92class PerRequestRootPathMiddleware: 

93 """Sets ``scope["root_path"]`` to the configured prefix matching the 

94 request path on a whole-segment boundary. ``scope["path"]`` is left 

95 untouched (Starlette strips ``root_path`` at route-match time). Must be 

96 the outermost middleware so inner middlewares and the router see the 

97 resolved value; a matched prefix overrides a scalar ``SERVER_ROOT_PATH`` 

98 for that request. 

99 """ 

100 

101 def __init__(self, app: ASGIApp, root_paths: Sequence[str]) -> None: 

102 self.app = app 

103 self.root_paths: Final = normalize_root_paths(root_paths) 

104 

105 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: 

106 if scope["type"] in ("http", "websocket"): 

107 path: Final = scope.get("path", "") 

108 for prefix in self.root_paths: 

109 if path == prefix or path.startswith(prefix + "/"): 

110 scope["root_path"] = prefix # rebind-ok: ASGI middleware contract; Router and base_url read it 

111 break 

112 # Stash the effective root_path (matched prefix, or the scope's 

113 # existing value when nothing matched — i.e. FastAPI's scalar 

114 # SERVER_ROOT_PATH) so code that emits URLs off the request path 

115 # picks the same prefix the router will resolve the request under. 

116 token: Final = _request_root_path_var.set(str(scope.get("root_path", ""))) 

117 try: 

118 await self.app(scope, receive, send) 

119 finally: 

120 _request_root_path_var.reset(token) 

121 return 

122 await self.app(scope, receive, send)