fix(proxy): gate delegated audit attribution

litellm/proxy/management_helpers/audit_logs.py

@@ -21,6 +21,17 @@ from litellm.proxy._types import (
 from litellm.types.utils import StandardAuditLogPayload
 
 _audit_log_callback_cache: Dict[str, CustomLogger] = {}
+ALLOW_LITELLM_CHANGED_BY_HEADER_METADATA_KEY = "allow_litellm_changed_by_header"
+
+
+def _allows_litellm_changed_by_header(user_api_key_dict: UserAPIKeyAuth) -> bool:
+    for admin_metadata in (user_api_key_dict.metadata, user_api_key_dict.team_metadata):
+        if (
+            isinstance(admin_metadata, dict)
+            and admin_metadata.get(ALLOW_LITELLM_CHANGED_BY_HEADER_METADATA_KEY) is True
+        ):
+            return True
+    return False
 
 
 def get_audit_log_changed_by(
@@ -29,6 +40,8 @@ def get_audit_log_changed_by(
     user_api_key_dict: UserAPIKeyAuth,
     litellm_proxy_admin_name: Optional[str],
 ) -> Optional[str]:
+    if litellm_changed_by and _allows_litellm_changed_by_header(user_api_key_dict):
+        return litellm_changed_by
     return user_api_key_dict.user_id or litellm_changed_by or litellm_proxy_admin_name
 
 

tests/proxy_unit_tests/test_audit_logs_proxy.py

@@ -73,6 +73,23 @@ def test_get_audit_log_changed_by_prefers_authenticated_user():
     )
 
 
+def test_get_audit_log_changed_by_honors_header_with_admin_opt_in():
+    user_api_key_dict = UserAPIKeyAuth(
+        api_key="test-key",
+        user_id="service-account",
+        metadata={"allow_litellm_changed_by_header": True},
+    )
+
+    assert (
+        get_audit_log_changed_by(
+            litellm_changed_by="delegated-user",
+            user_api_key_dict=user_api_key_dict,
+            litellm_proxy_admin_name="proxy-admin",
+        )
+        == "delegated-user"
+    )
+
+
 def test_get_audit_log_changed_by_falls_back_to_header_when_user_id_missing():
     user_api_key_dict = UserAPIKeyAuth(api_key="test-key")