Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/management_helpers/audit_logs.py: 31%

1""" 

2Functions to create audit logs for LiteLLM Proxy 

3""" 

4 

5import asyncio 

6import json 

7from datetime import datetime, timezone 

8from typing import Final 

9 

10import litellm 

11from litellm._logging import verbose_proxy_logger 

12from litellm._uuid import uuid 

13from litellm.integrations.custom_logger import CustomLogger 

14from litellm.proxy._types import ( 

15 AUDIT_ACTIONS, 

16 LiteLLM_AuditLogs, 

17 LitellmTableNames, 

18 UserAPIKeyAuth, 

19) 

20from litellm.repositories.table_repositories import AuditLogRepository 

21from litellm.types.utils import StandardAuditLogPayload 

22 

23_audit_log_callback_cache: Final[dict[str, CustomLogger]] = {} 

24ALLOW_LITELLM_CHANGED_BY_HEADER_METADATA_KEY: Final = "allow_litellm_changed_by_header" 

25 

26 

27def is_audit_logging_enabled(store_audit_logs: bool | None = None) -> bool: 

28 from litellm.secret_managers.main import get_secret_bool 

29 

30 configured_value: Final[bool | None] = litellm.store_audit_logs if store_audit_logs is None else store_audit_logs 

31 if configured_value is not None: 31 ↛ 32line 31 didn't jump to line 32 because the condition on line 31 was never true

32 return configured_value 

33 

34 environment_value: Final[bool | None] = get_secret_bool("LITELLM_STORE_AUDIT_LOGS") 

35 if environment_value is not None: 35 ↛ 36line 35 didn't jump to line 36 because the condition on line 35 was never true

36 return environment_value 

37 

38 from litellm.proxy.proxy_server import premium_user 

39 

40 return premium_user is True 

41 

42 

43def _allows_litellm_changed_by_header(user_api_key_dict: UserAPIKeyAuth) -> bool: 

44 for admin_metadata in (user_api_key_dict.metadata, user_api_key_dict.team_metadata): 

45 if ( 45 ↛ 49line 45 didn't jump to line 49 because the condition on line 45 was never true

46 isinstance(admin_metadata, dict) 

47 and admin_metadata.get(ALLOW_LITELLM_CHANGED_BY_HEADER_METADATA_KEY) is True 

48 ): 

49 return True 

50 return False 

51 

52 

53def get_audit_log_changed_by( 

54 *, 

55 litellm_changed_by: str | None, 

56 user_api_key_dict: UserAPIKeyAuth, 

57 litellm_proxy_admin_name: str | None, 

58) -> str | None: 

59 if litellm_changed_by and _allows_litellm_changed_by_header(user_api_key_dict): 59 ↛ 60line 59 didn't jump to line 60 because the condition on line 59 was never true

60 return litellm_changed_by 

61 return user_api_key_dict.user_id or litellm_proxy_admin_name 

62 

63 

64def _resolve_audit_log_callback(name: str) -> CustomLogger | None: 

65 """Resolve a string callback name to a CustomLogger instance, with caching. 

66 

67 For "s3_v2" with `litellm.s3_audit_callback_params` set, constructs a 

68 dedicated `S3Logger` so audit logs can target a different bucket than the 

69 normal-log singleton served by `_init_custom_logger_compatible_class`. 

70 """ 

71 if name in _audit_log_callback_cache: 

72 return _audit_log_callback_cache[name] 

73 

74 instance: CustomLogger | None 

75 if name == "s3_v2" and getattr(litellm, "s3_audit_callback_params", None) is not None: 

76 from litellm.integrations.s3_v2 import S3Logger as S3V2Logger 

77 

78 instance = S3V2Logger(s3_callback_params_override=litellm.s3_audit_callback_params) 

79 else: 

80 from litellm.litellm_core_utils.litellm_logging import ( 

81 _init_custom_logger_compatible_class, 

82 ) 

83 

84 instance = _init_custom_logger_compatible_class( 

85 logging_integration=name, 

86 internal_usage_cache=None, 

87 llm_router=None, 

88 ) 

89 

90 if instance is not None: 

91 _audit_log_callback_cache[name] = instance 

92 return instance 

93 

94 

95def reset_audit_log_callback_cache() -> None: 

96 """Clear cached audit-log callback instances. Call on config reload.""" 

97 _audit_log_callback_cache.clear() 

98 

99 

100def _build_audit_log_payload( 

101 request_data: LiteLLM_AuditLogs, 

102) -> StandardAuditLogPayload: 

103 """Convert LiteLLM_AuditLogs to StandardAuditLogPayload for callback dispatch.""" 

104 updated_at = "" 

105 if request_data.updated_at is not None: 

106 updated_at = request_data.updated_at.isoformat() 

107 

108 table_name_str: Final[str] = ( 

109 request_data.table_name.value 

110 if isinstance(request_data.table_name, LitellmTableNames) 

111 else str(request_data.table_name) 

112 ) 

113 

114 return StandardAuditLogPayload( 

115 id=request_data.id, 

116 updated_at=updated_at, 

117 changed_by=request_data.changed_by or "", 

118 changed_by_api_key=request_data.changed_by_api_key or "", 

119 action=request_data.action, 

120 table_name=table_name_str, 

121 object_id=request_data.object_id, 

122 before_value=request_data.before_value, 

123 updated_values=request_data.updated_values, 

124 ) 

125 

126 

127def _audit_log_task_done_callback(task: asyncio.Task) -> None: 

128 """Log exceptions from audit log callback tasks so they don't slip through silently.""" 

129 try: 

130 exc: Final = task.exception() 

131 except asyncio.CancelledError: 

132 return 

133 if exc is not None: 

134 verbose_proxy_logger.error("Audit log callback task failed: %s", exc, exc_info=exc) 

135 

136 

137async def _dispatch_audit_log_to_callbacks( 

138 request_data: LiteLLM_AuditLogs, 

139) -> None: 

140 """Dispatch audit log to all registered audit_log_callbacks.""" 

141 if not litellm.audit_log_callbacks: 

142 return 

143 

144 payload: Final = _build_audit_log_payload(request_data) 

145 

146 for callback in litellm.audit_log_callbacks: 

147 try: 

148 resolved: CustomLogger | None = callback if isinstance(callback, CustomLogger) else None 

149 if isinstance(callback, str): 

150 resolved = _resolve_audit_log_callback(callback) 

151 if resolved is None: 

152 verbose_proxy_logger.warning("Could not resolve audit log callback: %s", callback) 

153 continue 

154 

155 if isinstance(resolved, CustomLogger): 

156 task = asyncio.create_task(resolved.async_log_audit_log_event(payload)) 

157 task.add_done_callback(_audit_log_task_done_callback) 

158 except Exception as e: 

159 verbose_proxy_logger.error("Failed dispatching audit log to callback: %s", e) 

160 

161 

162async def create_object_audit_log( 

163 object_id: str, 

164 action: AUDIT_ACTIONS, 

165 litellm_changed_by: str | None, 

166 user_api_key_dict: UserAPIKeyAuth, 

167 litellm_proxy_admin_name: str | None, 

168 table_name: LitellmTableNames, 

169 before_value: str | None = None, 

170 after_value: str | None = None, 

171): 

172 """ 

173 Create an audit log for an internal user. 

174 

175 Parameters: 

176 - user_id: str - The id of the user to create the audit log for. 

177 - action: AUDIT_ACTIONS - The action to create the audit log for. 

178 - user_row: LiteLLM_UserTable - The user row to create the audit log for. 

179 - litellm_changed_by: Optional[str] - The user id of the user who is changing the user. 

180 - user_api_key_dict: UserAPIKeyAuth - The user api key dictionary. 

181 - litellm_proxy_admin_name: Optional[str] - The name of the proxy admin. 

182 """ 

183 if not is_audit_logging_enabled(): 183 ↛ 186line 183 didn't jump to line 186 because the condition on line 183 was always true

184 return 

185 

186 _changed_by: Final = get_audit_log_changed_by( 

187 litellm_changed_by=litellm_changed_by, 

188 user_api_key_dict=user_api_key_dict, 

189 litellm_proxy_admin_name=litellm_proxy_admin_name, 

190 ) 

191 

192 await create_audit_log_for_update( 

193 request_data=LiteLLM_AuditLogs( 

194 id=str(uuid.uuid4()), 

195 updated_at=datetime.now(timezone.utc), 

196 changed_by=_changed_by, 

197 changed_by_api_key=user_api_key_dict.api_key, 

198 table_name=table_name, 

199 object_id=object_id, 

200 action=action, 

201 updated_values=after_value, 

202 before_value=before_value, 

203 ) 

204 ) 

205 

206 

207async def create_audit_log_for_update(request_data: LiteLLM_AuditLogs): 

208 """ 

209 Create an audit log for an object. 

210 """ 

211 if not is_audit_logging_enabled(): 

212 return 

213 

214 from litellm.proxy.proxy_server import premium_user, prisma_client 

215 

216 if premium_user is not True: 

217 return 

218 

219 verbose_proxy_logger.debug("creating audit log for %s", request_data) 

220 

221 if isinstance(request_data.updated_values, dict): 

222 request_data.updated_values = json.dumps(request_data.updated_values) 

223 

224 if isinstance(request_data.before_value, dict): 

225 request_data.before_value = json.dumps(request_data.before_value) 

226 

227 # Dispatch to external audit log callbacks regardless of DB availability 

228 await _dispatch_audit_log_to_callbacks(request_data) 

229 

230 if prisma_client is None: 

231 verbose_proxy_logger.error("prisma_client is None, cannot write audit log to DB") 

232 return 

233 

234 _request_data: Final = request_data.model_dump(exclude_none=True) 

235 

236 try: 

237 await AuditLogRepository(prisma_client).table.create( 

238 data={ 

239 **_request_data, 

240 } 

241 ) 

242 except Exception as e: 

243 # [Non-Blocking Exception. Do not allow blocking LLM API call] 

244 verbose_proxy_logger.error("Failed Creating audit log %s", e)