Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/plugin_routes.py: 36%

1""" 

2Plugin proxy routes for litellm. 

3 

4Enables external services to register as plugins and be proxied through 

5the litellm proxy server. 

6 

7Config (in litellm config.yaml general_settings): 

8 plugins: 

9 - name: my-plugin 

10 url: "http://localhost:3210" 

11 display_name: "My Plugin" 

12 plugin_key: "sk-..." # optional: plugin's own auth key 

13 

14Plugin iframe auth: 

15 The UI calls GET /api/plugins/auth-token to receive a short-lived identity 

16 claim ({user_id, user_role, plugin, exp}) encrypted with a per-plugin key 

17 derived as HMAC-SHA256(LITELLM_SALT_KEY, plugin_name). The claim carries no 

18 litellm bearer token, so a compromised plugin learns only the caller's 

19 identity, never their credential. LITELLM_SALT_KEY itself is never shared 

20 with plugins — each plugin holds only its own derived key. 

21""" 

22 

23import base64 

24import hashlib 

25import hmac as _hmac 

26import json 

27import os 

28import time 

29from collections.abc import Mapping 

30from typing import Final 

31 

32from cryptography.fernet import Fernet, InvalidToken 

33from fastapi import APIRouter, Depends, HTTPException, Request, Response 

34 

35from litellm.llms.custom_httpx.http_handler import get_async_httpx_client 

36from litellm.proxy._types import PluginConfig, SpecialHeaders, UserAPIKeyAuth 

37from litellm.proxy.auth.user_api_key_auth import user_api_key_auth 

38from litellm.types.llms.custom_http import httpxSpecialProvider 

39 

40router: Final = APIRouter() 

41 

42# Hop-by-hop headers (RFC 7230) and the litellm session cookie — never forwarded 

43# to a plugin backend. Credential headers are added on top per-request from the 

44# canonical SpecialHeaders set so the plugin only ever authenticates via its own 

45# injected plugin_key. 

46_HOP_BY_HOP_STRIP: Final = frozenset( 

47 { 

48 "host", 

49 "connection", 

50 "transfer-encoding", 

51 "te", 

52 "trailers", 

53 "upgrade", 

54 "cookie", 

55 } 

56) 

57 

58 

59def _configured_key_header_names() -> frozenset[str]: 

60 """The lowercased general_settings.litellm_key_header_name, if configured. 

61 

62 Read live from the proxy module (not import-time) so a custom key header set 

63 via config is honoured without a restart. Returns empty when unset. 

64 """ 

65 try: 

66 from litellm.proxy import proxy_server 

67 except Exception: 

68 return frozenset() 

69 general_settings: Final = getattr(proxy_server, "general_settings", None) 

70 if not isinstance(general_settings, Mapping): 

71 return frozenset() 

72 name: Final[object] = general_settings.get("litellm_key_header_name") 

73 return frozenset({name.lower()}) if isinstance(name, str) and name else frozenset() 

74 

75 

76def _request_strip_headers() -> frozenset[str]: 

77 """Headers to drop before forwarding a request to a plugin backend. 

78 

79 Every header user_api_key_auth accepts as a litellm credential is stripped — 

80 Authorization, x-api-key, API-Key, x-goog-api-key, Ocp-Apim-Subscription-Key, 

81 x-litellm-api-key, and any configured custom key header — so a plugin can 

82 never be handed the caller's live litellm key (confused-deputy escalation). 

83 """ 

84 return _HOP_BY_HOP_STRIP | SpecialHeaders.litellm_credential_header_names() | _configured_key_header_names() 

85 

86 

87# Headers to strip from plugin RESPONSES before returning to the browser. 

88# httpx already decompresses and de-chunks the body, so forwarding the wire 

89# encoding headers causes clients to attempt double-decompression (garbage) or 

90# incorrect length checks. set-cookie is removed so plugins cannot overwrite 

91# litellm session cookies. 

92_RESPONSE_STRIP: Final = { 

93 "content-encoding", 

94 "transfer-encoding", 

95 "content-length", 

96 "set-cookie", 

97} 

98 

99 

100def _safe_response_headers(raw: "Mapping[str, str]") -> dict[str, str]: 

101 """Strip wire-encoding/cookie headers and force proxied responses inert. 

102 

103 Plugin-controlled bytes are served from the litellm dashboard origin, so a 

104 compromised plugin could return an HTML/JS document that executes with the 

105 admin's session against same-origin management APIs. A sandbox CSP forces 

106 the response into an opaque origin with scripts disabled, and nosniff stops 

107 content-type confusion from re-enabling execution. Both are set last so a 

108 plugin cannot override them with its own headers. 

109 """ 

110 return { 

111 **{k: v for k, v in raw.items() if k.lower() not in _RESPONSE_STRIP}, 

112 "content-security-policy": "sandbox", 

113 "x-content-type-options": "nosniff", 

114 } 

115 

116 

117# In-memory plugin registry — populated from general_settings at startup 

118_plugin_registry: Final[dict[str, PluginConfig]] = {} 

119 

120 

121# --------------------------------------------------------------------------- 

122# Key derivation — audience-scoped per plugin so compromising one plugin 

123# cannot be used to forge claims for another. LITELLM_SALT_KEY is NEVER 

124# shared with plugins; each plugin only receives a key derived from 

125# HMAC(LITELLM_SALT_KEY, plugin_name) which reveals nothing about the master. 

126# --------------------------------------------------------------------------- 

127def _plugin_fernet(plugin_name: str) -> Fernet: 

128 """Return a Fernet cipher whose key is scoped to a specific plugin. 

129 

130 Key material: HMAC-SHA256(LITELLM_SALT_KEY, plugin_name). 

131 A plugin possessing its own key cannot derive the master salt or 

132 forge claims intended for a different plugin. 

133 """ 

134 salt: Final = os.getenv("LITELLM_SALT_KEY", "").encode() 

135 derived: Final = _hmac.new(salt, plugin_name.encode(), hashlib.sha256).digest() 

136 return Fernet(base64.urlsafe_b64encode(derived)) 

137 

138 

139_CLAIM_TTL_SECONDS: Final = 30 # identity claims expire after 30 s 

140 

141 

142def issue_plugin_session_claim(plugin_name: str, user_id: str | None, user_role: str | None) -> str: 

143 """Issue a short-lived, audience-scoped identity claim for the plugin. 

144 

145 The claim contains {user_id, user_role, plugin, exp}. Crucially it 

146 contains NO litellm bearer token — the plugin can only derive the 

147 caller's identity, not act as them against the proxy. 

148 """ 

149 claim: Final = { 

150 "plugin": plugin_name, 

151 "user_id": user_id or "", 

152 "user_role": user_role or "", 

153 "exp": int(time.time()) + _CLAIM_TTL_SECONDS, 

154 } 

155 return _plugin_fernet(plugin_name).encrypt(json.dumps(claim).encode()).decode() 

156 

157 

158def verify_plugin_session_claim(plugin_name: str, ciphertext: str) -> dict: 

159 """Verify and decode a plugin session claim. 

160 

161 Raises ValueError if the HMAC is invalid, the audience is wrong, or 

162 the claim is expired. Returns the decoded claim dict on success. 

163 """ 

164 try: 

165 raw: Final = _plugin_fernet(plugin_name).decrypt(ciphertext.encode(), ttl=_CLAIM_TTL_SECONDS) 

166 claim: Final = json.loads(raw) 

167 except (InvalidToken, Exception) as exc: 

168 raise ValueError("Invalid, tampered, or expired plugin session claim") from exc 

169 

170 if claim.get("plugin") != plugin_name: 

171 raise ValueError("Plugin claim audience mismatch") 

172 if int(claim.get("exp", 0)) < int(time.time()): 

173 raise ValueError("Plugin session claim expired") 

174 return claim 

175 

176 

177# --------------------------------------------------------------------------- 

178# Config 

179# --------------------------------------------------------------------------- 

180def register_plugins_from_config(general_settings: dict[str, object]) -> None: 

181 """Replace the plugin registry from general_settings. 

182 

183 Replaces (not merges) so plugins removed from config are immediately 

184 unreachable without requiring a process restart. 

185 """ 

186 raw: Final = general_settings.get("plugins") 

187 entries: Final[list[object]] = raw if isinstance(raw, list) else [] 

188 new_registry: Final = {p.name: p for p in (PluginConfig.model_validate(entry) for entry in entries)} 

189 _plugin_registry.clear() 

190 _plugin_registry.update(new_registry) 

191 

192 

193# --------------------------------------------------------------------------- 

194# Routes 

195# --------------------------------------------------------------------------- 

196@router.get("/api/plugins", tags=["plugins"]) 

197async def list_plugins( 

198 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), 

199) -> list[dict[str, str]]: 

200 """Return registered plugins for authenticated UI callers. 

201 

202 plugin_key is never returned — the browser never needs it (the proxy injects 

203 it server-side from the registry), and exposing it here would leak the 

204 credential into React state and DevTools. Admin key management goes through 

205 the redacted /config/field/info path instead. 

206 """ 

207 return [ 

208 { 

209 "name": plugin.name, 

210 "display_name": plugin.display_name or plugin.name, 

211 "url": plugin.url, 

212 } 

213 for plugin in _plugin_registry.values() 

214 ] 

215 

216 

217@router.get("/api/plugins/auth-token", tags=["plugins"]) 

218async def plugin_auth_token( 

219 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), 

220 plugin_name: str = "litellm-platform-plugin", 

221) -> dict: 

222 """Issue a short-lived, audience-scoped plugin session claim. 

223 

224 The claim contains {user_id, user_role, plugin, exp}. It does NOT 

225 contain the caller's litellm bearer token — a compromised plugin can 

226 only learn the caller's identity, not impersonate them against the proxy. 

227 

228 Encrypted with a key derived from HMAC(LITELLM_SALT_KEY, plugin_name), 

229 so each plugin holds only its own key and cannot forge claims for others. 

230 

231 Requires LITELLM_SALT_KEY to be set; returns 503 otherwise. 

232 """ 

233 if not os.getenv("LITELLM_SALT_KEY"): 233 ↛ 234line 233 didn't jump to line 234 because the condition on line 233 was never true

234 raise HTTPException( 

235 status_code=503, 

236 detail="LITELLM_SALT_KEY is not configured; plugin iframe auth unavailable.", 

237 ) 

238 if plugin_name not in _plugin_registry: 238 ↛ 240line 238 didn't jump to line 240 because the condition on line 238 was always true

239 raise HTTPException(status_code=404, detail=f"Plugin '{plugin_name}' is not registered.") 

240 user_id: Final = getattr(user_api_key_dict, "user_id", None) 

241 user_role: Final = getattr(user_api_key_dict, "user_role", None) 

242 return {"session_claim": issue_plugin_session_claim(plugin_name, user_id, user_role)} 

243 

244 

245@router.api_route( 

246 "/plugin-proxy/{plugin_name}/{path:path}", 

247 methods=["GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS", "HEAD"], 

248 tags=["plugins"], 

249 include_in_schema=False, 

250) 

251async def plugin_proxy( 

252 plugin_name: str, 

253 path: str, 

254 request: Request, 

255 user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), 

256) -> Response: 

257 """Authenticated reverse-proxy to a registered plugin backend. 

258 

259 Restricted to proxy_admin callers — the shared plugin_key must not be 

260 usable as a confused-deputy credential by regular users. Plugin UIs 

261 talk to the plugin service directly via the iframe; this route is for 

262 administrative and server-to-server access only. 

263 

264 The caller's litellm credential is stripped and replaced with the 

265 plugin's own plugin_key so plugins never receive a live litellm API key. 

266 """ 

267 if getattr(user_api_key_dict, "user_role", None) != "proxy_admin": 

268 return Response( 

269 content="Plugin proxy access requires proxy_admin role.", 

270 status_code=403, 

271 ) 

272 

273 plugin: Final = _plugin_registry.get(plugin_name) 

274 if not plugin: 

275 return Response( 

276 content=f"Plugin '{plugin_name}' not registered", 

277 status_code=404, 

278 ) 

279 

280 target_url = f"{plugin.url.rstrip('/')}/{path}" 

281 query: Final = request.url.query 

282 if query: 

283 target_url = f"{target_url}?{query}" 

284 

285 body: Final = await request.body() 

286 

287 # Strip caller credentials and hop-by-hop headers from forwarded request 

288 strip: Final = _request_strip_headers() 

289 forward_headers: Final = {k: v for k, v in request.headers.items() if k.lower() not in strip} 

290 

291 # Inject plugin's own credential as upstream auth (if configured) 

292 plugin_key: Final = plugin.plugin_key 

293 if plugin_key: 

294 forward_headers["authorization"] = f"Bearer {plugin_key}" 

295 

296 # Forward caller identity so the plugin can enforce its own access control. 

297 # The plugin MUST NOT trust these as credentials — they are informational. 

298 # The plugin_key above is the only authentication mechanism. 

299 user_id: Final = getattr(user_api_key_dict, "user_id", None) 

300 user_role: Final = getattr(user_api_key_dict, "user_role", None) 

301 if user_id: 

302 forward_headers["x-litellm-user-id"] = str(user_id) 

303 if user_role: 

304 forward_headers["x-litellm-user-role"] = str(user_role) 

305 

306 handler: Final = get_async_httpx_client(llm_provider=httpxSpecialProvider.PassThroughEndpoint) 

307 try: 

308 req: Final = handler.client.build_request( 

309 method=request.method, 

310 url=target_url, 

311 headers=forward_headers, 

312 content=body, 

313 ) 

314 # Do not follow redirects — a redirect to an internal URL would allow 

315 # the plugin to SSRF the proxy into fetching arbitrary internal services. 

316 resp: Final = await handler.client.send(req, follow_redirects=False) 

317 except Exception: 

318 return Response( 

319 content=f"Cannot connect to plugin '{plugin_name}' at {plugin.url}", 

320 status_code=502, 

321 ) 

322 

323 return Response( 

324 content=resp.content, 

325 status_code=resp.status_code, 

326 headers=_safe_response_headers(resp.headers), 

327 )