Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/common_utils/callback_config_validation.py: 21%

1"""Save-time validation of team/key logging configs the runtime cannot honor. 

2 

3Team callbacks arrive as a single ``AddTeamCallback``, key callbacks arrive as a 

4``logging`` list inside the key metadata, so both shapes funnel into the same 

5per-integration checks here. 

6""" 

7 

8import math 

9from collections.abc import Mapping, Sequence 

10from types import MappingProxyType 

11from typing import Final 

12 

13_NEWRELIC_CALLBACK: Final = "newrelic" 

14_NEWRELIC_VAR_PREFIX: Final = "newrelic_" 

15_LANGFUSE_OTEL_CALLBACK: Final = "langfuse_otel" 

16_LANGFUSE_SPAN_SCOPE_VAR: Final = "langfuse_span_scope" 

17_ARIZE_CALLBACK: Final = "arize" 

18_ARIZE_SAMPLING_RATE_VARS: Final[frozenset[str]] = frozenset( 

19 {"arize_success_sampling_rate", "arize_error_sampling_rate"} 

20) 

21 

22 

23def callback_config_error(callback_name: str | None, callback_vars: Mapping[str, str] | None) -> str | None: 

24 if not callback_vars: 24 ↛ 26line 24 didn't jump to line 26 because the condition on line 24 was always true

25 return None 

26 arize_error: Final = _arize_sampling_rate_error(callback_name, callback_vars) 

27 if arize_error is not None: 

28 return arize_error 

29 langfuse_error: Final = _langfuse_environment_error(callback_vars) or _langfuse_span_scope_error( 

30 callback_name, callback_vars 

31 ) 

32 if langfuse_error is not None: 

33 return langfuse_error 

34 if callback_name != _NEWRELIC_CALLBACK: 

35 return None 

36 return _newrelic_config_error(callback_vars) 

37 

38 

39def _langfuse_environment_error(callback_vars: Mapping[str, str]) -> str | None: 

40 """Reject langfuse_environment values Langfuse ingestion would drop. 

41 

42 Accepting an invalid value here would 200 the config write and then 

43 silently lose every trace for that key/team at request time. 

44 """ 

45 value: Final = callback_vars.get("langfuse_environment") 

46 if value is None: 

47 return None 

48 from litellm.litellm_core_utils.initialize_dynamic_callback_params import ( 

49 validate_langfuse_environment_value, 

50 ) 

51 

52 try: 

53 validate_langfuse_environment_value(value) 

54 except ValueError as e: 

55 return str(e) 

56 return None 

57 

58 

59def _langfuse_span_scope_error(callback_name: str | None, callback_vars: Mapping[str, str]) -> str | None: 

60 value: Final = callback_vars.get(_LANGFUSE_SPAN_SCOPE_VAR) 

61 if value is None: 

62 return None 

63 if callback_name != _LANGFUSE_OTEL_CALLBACK: 

64 return ( 

65 f"{_LANGFUSE_SPAN_SCOPE_VAR} applies to the {_LANGFUSE_OTEL_CALLBACK} callback only, not {callback_name!r}" 

66 ) 

67 from litellm.litellm_core_utils.initialize_dynamic_callback_params import ( 

68 validate_langfuse_span_scope_value, 

69 ) 

70 

71 try: 

72 validate_langfuse_span_scope_value(value) 

73 except ValueError as e: 

74 return str(e) 

75 return None 

76 

77 

78# Which credential family a dynamic variable belongs to. The families are the 

79# integrations that share one account: every langfuse_* variable configures the 

80# same Langfuse project whether it rides the classic callback or the OTel one, 

81# and every dd_* variable configures the same Datadog account. 

82_VAR_FAMILIES: Final[Mapping[str, str]] = MappingProxyType( 

83 { 

84 "arize_": "Arize", 

85 "dd_": "Datadog", 

86 "gcs_": "GCS", 

87 "humanloop_": "Humanloop", 

88 "langfuse_": "Langfuse", 

89 "langsmith_": "LangSmith", 

90 "newrelic_": "New Relic", 

91 "posthog_": "PostHog", 

92 "wandb_": "Weights & Biases", 

93 "weave_": "Weights & Biases", 

94 } 

95) 

96 

97_FAMILY_OPTION_VARS: Final[frozenset[str]] = frozenset({_LANGFUSE_SPAN_SCOPE_VAR, *_ARIZE_SAMPLING_RATE_VARS}) 

98 

99 

100def _family_of(var: str) -> str | None: 

101 """The credential family ``var`` configures, or ``None`` if it configures none. 

102 

103 ``turn_off_message_logging`` and friends belong to no backend, so they carry 

104 no credentials anyone could redirect. ``langfuse_span_scope`` shares the Langfuse 

105 prefix but is a fixed enum choosing what the family exports, not where to. 

106 """ 

107 if var in _FAMILY_OPTION_VARS: 

108 return None 

109 return next((family for prefix, family in _VAR_FAMILIES.items() if var.startswith(prefix)), None) 

110 

111 

112def cross_entry_family_error( 

113 callback_vars: Mapping[str, str] | None, 

114 stored_vars_by_entry: Sequence[Mapping[str, str]], 

115) -> str | None: 

116 """Reject an entry that changes what a family another entry holds resolves to. 

117 

118 Every stored entry's variables are flattened into one dict before a request 

119 reads them, and the flattened dict is what the exporter authenticates and 

120 addresses with. So an entry naming only a destination is enough to redirect 

121 credentials that were written somewhere else: a host on a second entry pairs 

122 with the key from the first, and the request carries that key to the new 

123 host. 

124 

125 Two rules together keep the flattened dict out of the caller's hands. A 

126 variable the family already configures has to keep the value it has, so 

127 nothing already in use can be moved. A variable the family does not yet 

128 configure may only carry a value the family already holds, which is what lets 

129 the same credential go in under its other spelling (``langfuse_secret`` and 

130 ``langfuse_secret_key`` are one key) without anything here having to list the 

131 spellings. Between them, no value the caller chose can enter the family, and 

132 repeating the family as it stands is still allowed -- that is how one 

133 integration gets registered for both the success and the failure event. 

134 

135 A team admin who does want to move a family deletes the entry holding it 

136 first, which reveals nothing. 

137 

138 Only the writers this endpoint newly admits are held to this, because a proxy 

139 admin already holds every credential the proxy has. 

140 

141 ``stored_vars_by_entry`` has to arrive decrypted; the credential values are 

142 encrypted at rest and ciphertext never equals the plaintext coming in. 

143 """ 

144 if not callback_vars: 

145 return None 

146 stored_by_var: Final = { 

147 var: value for entry in stored_vars_by_entry for var, value in entry.items() if _family_of(var) is not None 

148 } 

149 family_values: Final = frozenset( 

150 (family, value) 

151 for entry in stored_vars_by_entry 

152 for var, value in entry.items() 

153 if (family := _family_of(var)) is not None 

154 ) 

155 held_families: Final = frozenset(family for family, _ in family_values) 

156 return next( 

157 ( 

158 f"{family} is already configured by another callback entry on this team. " 

159 f"Remove that entry before setting {var} here." 

160 for var, value, family in ((v, callback_vars[v], _family_of(v)) for v in callback_vars) 

161 if family in held_families 

162 and (stored_by_var[var] != value if var in stored_by_var else (family, value) not in family_values) 

163 ), 

164 None, 

165 ) 

166 

167 

168def conflicting_span_scope_error( 

169 callback_vars: Mapping[str, str] | None, 

170 stored_vars_by_entry: Sequence[Mapping[str, str]], 

171) -> str | None: 

172 incoming: Final = None if callback_vars is None else callback_vars.get(_LANGFUSE_SPAN_SCOPE_VAR) 

173 if incoming is None: 173 ↛ 175line 173 didn't jump to line 175 because the condition on line 173 was always true

174 return None 

175 return next( 

176 ( 

177 f"{_LANGFUSE_SPAN_SCOPE_VAR} is already set to {stored!r} by another callback entry. " 

178 f"Every entry shares one scope: remove that entry or send the same value." 

179 for entry in stored_vars_by_entry 

180 if (stored := entry.get(_LANGFUSE_SPAN_SCOPE_VAR)) not in (None, incoming) 

181 ), 

182 None, 

183 ) 

184 

185 

186def logging_metadata_config_error(metadata: Mapping[str, object] | None) -> str | None: 

187 """Validate every ``logging`` entry of a team/key metadata payload.""" 

188 if not metadata: 188 ↛ 190line 188 didn't jump to line 190 because the condition on line 188 was always true

189 return None 

190 entries: Final = metadata.get("logging") 

191 if not isinstance(entries, Sequence) or isinstance(entries, (str, bytes)): 

192 return None 

193 entry_vars: Final = tuple(_entry_callback_vars(entry) for entry in entries) 

194 return next( 

195 ( 

196 error 

197 for error in ( 

198 *(_logging_entry_error(entry) for entry in entries), 

199 *(conflicting_span_scope_error(entry_vars[i], entry_vars[:i]) for i in range(len(entry_vars))), 

200 ) 

201 if error is not None 

202 ), 

203 None, 

204 ) 

205 

206 

207def _entry_callback_vars(entry: object) -> Mapping[str, str]: 

208 callback_vars: Final = entry.get("callback_vars") if isinstance(entry, Mapping) else None 

209 if not isinstance(callback_vars, Mapping): 

210 return MappingProxyType({}) 

211 return MappingProxyType({str(key): str(value) for key, value in callback_vars.items()}) 

212 

213 

214def _logging_entry_error(entry: object) -> str | None: 

215 if not isinstance(entry, Mapping): 

216 return None 

217 callback_name: Final = entry.get("callback_name") 

218 if not isinstance(callback_name, str) or not isinstance(entry.get("callback_vars"), Mapping): 

219 return None 

220 return callback_config_error(callback_name, _entry_callback_vars(entry)) 

221 

222 

223def _arize_sampling_rate_error(callback_name: str | None, callback_vars: Mapping[str, str]) -> str | None: 

224 for var in sorted(_ARIZE_SAMPLING_RATE_VARS): 

225 value = callback_vars.get(var) 

226 if value is None or value in ("", "None"): 

227 continue 

228 if callback_name != _ARIZE_CALLBACK: 

229 return f"{var} applies to the {_ARIZE_CALLBACK} callback only, not {callback_name!r}" 

230 try: 

231 rate = float(value) 

232 except (TypeError, ValueError): 

233 return f"{var} must be a number between 0.0 and 1.0 (inclusive), got {value!r}" 

234 if not math.isfinite(rate) or not 0.0 <= rate <= 1.0: 

235 return f"{var} must be a number between 0.0 and 1.0 (inclusive), got {value!r}" 

236 return None 

237 

238 

239def _newrelic_config_error(callback_vars: Mapping[str, str]) -> str | None: 

240 """Per-team New Relic routing runs on the OTel v2 path only. 

241 

242 Accepting the config with the flag off would silently ship the team's traffic 

243 through the operator's env-configured agent instead of the team's account. A 

244 region outside the fixed table, or a region without a key, would likewise be 

245 accepted and then silently ignored or misrouted at request time. 

246 """ 

247 if not any(key.startswith(_NEWRELIC_VAR_PREFIX) for key in callback_vars): 

248 return None 

249 

250 from litellm.integrations.otel.model.config import is_otel_v2_enabled 

251 from litellm.integrations.otel.presets.newrelic import NEWRELIC_OTLP_ENDPOINT_BY_REGION 

252 

253 if not is_otel_v2_enabled(): 

254 return "Per-team New Relic routing requires the proxy to run with LITELLM_OTEL_V2=true." 

255 

256 region: Final = callback_vars.get("newrelic_region") 

257 if region is not None and region.lower() not in NEWRELIC_OTLP_ENDPOINT_BY_REGION: 

258 return ( 

259 f"Unknown newrelic_region {region!r}. " 

260 f"Supported regions: {', '.join(sorted(NEWRELIC_OTLP_ENDPOINT_BY_REGION))}." 

261 ) 

262 

263 # ``callback_vars`` values are str()-coerced upstream, so a JSON ``null`` key 

264 # arrives as the literal ``"None"``; treat that and the empty string as absent. 

265 api_key: Final = callback_vars.get("newrelic_api_key") 

266 if region is not None and (not api_key or api_key == "None"): 

267 return "newrelic_region requires newrelic_api_key; the region rides the team's own key." 

268 return None