scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-07 02:59:05 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 103

fix: merge team metadata in admin helper, remove turn_off_message_logging from dynamic params

Browse source 
Include user_api_key_team_metadata alongside user_api_key_metadata in
_get_admin_metadata() so team-level guardrail settings are respected.
Key-level settings take precedence over team-level.

Remove turn_off_message_logging from _supported_callback_params so it
cannot be set via request metadata. Admin controls logging globally
or via key/team configuration.

Update tests to verify user-injected guardrail flags are ignored while
admin-configured flags are respected.
This commit is contained in:
user 2026-04-16 02:41:41 +00:00
parent 3cd5796fc7
commit 34e9be1ba7
No known key found for this signature in database
3 changed files with 53 additions and 73 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
litellm/integrations/custom_guardrail.py
View file

@ -257,9 +257,12 @@ class CustomGuardrail(CustomLogger):
@staticmethod @staticmethod
def _get_admin_metadata(data: dict) -> dict: def _get_admin_metadata(data: dict) -> dict:
"""Return the admin-configured key/team metadata from the request data.""" """Return merged admin-configured key and team metadata from the request data."""
metadata = data.get("litellm_metadata") or data.get("metadata", {}) metadata = data.get("litellm_metadata") or data.get("metadata", {})
return metadata.get("user_api_key_metadata") or {} team_meta = metadata.get("user_api_key_team_metadata") or {}
key_meta = metadata.get("user_api_key_metadata") or {}
# Key-level settings override team-level
return {**team_meta, **key_meta}
def get_disable_global_guardrail(self, data: dict) -> Optional[bool]: def get_disable_global_guardrail(self, data: dict) -> Optional[bool]:
""" """

1
litellm/litellm_core_utils/initialize_dynamic_callback_params.py
View file

@ -48,7 +48,6 @@ _supported_callback_params = [
"braintrust_host", "braintrust_host",
"slack_webhook_url", "slack_webhook_url",
"lunary_public_key", "lunary_public_key",
"turn_off_message_logging",
] ]

118
tests/test_litellm/integrations/test_custom_guardrail.py
View file

@ -173,17 +173,16 @@ class TestCustomGuardrailShouldRunGuardrail:
assert result is False assert result is False
def test_should_run_guardrail_with_disable_global_guardrail(self): def test_should_run_guardrail_with_disable_global_guardrail(self):
"""Test that disable_global_guardrail disables a global guardrail when set to True""" """Test that disable_global_guardrails only works from admin metadata"""
from litellm.types.guardrails import GuardrailEventHooks from litellm.types.guardrails import GuardrailEventHooks
# Create a guardrail with default_on=True (global guardrail)
custom_guardrail = CustomGuardrail( custom_guardrail = CustomGuardrail(
guardrail_name="global_guardrail", guardrail_name="global_guardrail",
default_on=True, default_on=True,
event_hook=GuardrailEventHooks.pre_call, event_hook=GuardrailEventHooks.pre_call,
) )
# Test 1: Global guardrail runs by default when default_on=True # Test 1: Global guardrail runs by default
data = { data = {
"model": "gpt-3.5-turbo", "model": "gpt-3.5-turbo",
"messages": [{"role": "user", "content": "test"}], "messages": [{"role": "user", "content": "test"}],
@ -193,7 +192,7 @@ class TestCustomGuardrailShouldRunGuardrail:
) )
assert result is True, "Global guardrail should run when default_on=True" assert result is True, "Global guardrail should run when default_on=True"
# Test 2: Global guardrail is disabled when disable_global_guardrail=True at root level # Test 2: User-injected disable at root level is IGNORED
data_with_disable_root = { data_with_disable_root = {
"model": "gpt-3.5-turbo", "model": "gpt-3.5-turbo",
"messages": [{"role": "user", "content": "test"}], "messages": [{"role": "user", "content": "test"}],
@ -203,23 +202,10 @@ class TestCustomGuardrailShouldRunGuardrail:
data=data_with_disable_root, event_type=GuardrailEventHooks.pre_call data=data_with_disable_root, event_type=GuardrailEventHooks.pre_call
) )
assert ( assert (
result is False result is True
), "Global guardrail should be disabled when disable_global_guardrail=True" ), "User-injected disable_global_guardrails should be ignored"
# Test 3: Global guardrail is disabled when disable_global_guardrail=True in litellm_metadata # Test 3: User-injected disable in metadata is IGNORED
data_with_disable_litellm = {
"model": "gpt-3.5-turbo",
"messages": [{"role": "user", "content": "test"}],
"litellm_metadata": {"disable_global_guardrails": True},
}
result = custom_guardrail.should_run_guardrail(
data=data_with_disable_litellm, event_type=GuardrailEventHooks.pre_call
)
assert (
result is False
), "Global guardrail should be disabled when disable_global_guardrail=True in litellm_metadata"
# Test 4: Global guardrail is disabled when disable_global_guardrail=True in metadata
data_with_disable_metadata = { data_with_disable_metadata = {
"model": "gpt-3.5-turbo", "model": "gpt-3.5-turbo",
"messages": [{"role": "user", "content": "test"}], "messages": [{"role": "user", "content": "test"}],
@ -228,25 +214,21 @@ class TestCustomGuardrailShouldRunGuardrail:
result = custom_guardrail.should_run_guardrail( result = custom_guardrail.should_run_guardrail(
data=data_with_disable_metadata, event_type=GuardrailEventHooks.pre_call data=data_with_disable_metadata, event_type=GuardrailEventHooks.pre_call
) )
assert ( assert result is True, "User-injected metadata disable should be ignored"
result is False
), "Global guardrail should be disabled when disable_global_guardrail=True in metadata"
# Test 5: Global guardrail runs when disable_global_guardrail=False # Test 4: Admin-configured disable via user_api_key_metadata IS respected
data_with_disable_false = { data_with_admin_disable = {
"model": "gpt-3.5-turbo", "model": "gpt-3.5-turbo",
"messages": [{"role": "user", "content": "test"}], "messages": [{"role": "user", "content": "test"}],
"disable_global_guardrails": False, "metadata": {"user_api_key_metadata": {"disable_global_guardrails": True}},
} }
result = custom_guardrail.should_run_guardrail( result = custom_guardrail.should_run_guardrail(
data=data_with_disable_false, event_type=GuardrailEventHooks.pre_call data=data_with_admin_disable, event_type=GuardrailEventHooks.pre_call
) )
assert ( assert result is False, "Admin-configured disable should be respected"
result is True
), "Global guardrail should still run when disable_global_guardrail=False"
def test_should_run_guardrail_with_opted_out_global_guardrails(self): def test_should_run_guardrail_with_opted_out_global_guardrails(self):
"""Test the per-guardrail opt-out list for global (default_on=True) guardrails""" """Test that per-guardrail opt-out only works from admin metadata"""
from litellm.types.guardrails import GuardrailEventHooks from litellm.types.guardrails import GuardrailEventHooks
custom_guardrail = CustomGuardrail( custom_guardrail = CustomGuardrail(
@ -255,7 +237,7 @@ class TestCustomGuardrailShouldRunGuardrail:
event_hook=GuardrailEventHooks.pre_call, event_hook=GuardrailEventHooks.pre_call,
) )
# Test 1: guardrail in the opt-out list at root level → skipped # Test 1: User-injected opt-out at root level is IGNORED
data_root = { data_root = {
"model": "gpt-3.5-turbo", "model": "gpt-3.5-turbo",
"messages": [{"role": "user", "content": "test"}], "messages": [{"role": "user", "content": "test"}],
@ -265,23 +247,10 @@ class TestCustomGuardrailShouldRunGuardrail:
custom_guardrail.should_run_guardrail( custom_guardrail.should_run_guardrail(
data=data_root, event_type=GuardrailEventHooks.pre_call data=data_root, event_type=GuardrailEventHooks.pre_call
) )
is False is True
) )
# Test 2: guardrail in the opt-out list inside litellm_metadata → skipped # Test 2: User-injected opt-out in metadata is IGNORED
data_litellm = {
"model": "gpt-3.5-turbo",
"messages": [{"role": "user", "content": "test"}],
"litellm_metadata": {"opted_out_global_guardrails": ["global_guardrail"]},
}
assert (
custom_guardrail.should_run_guardrail(
data=data_litellm, event_type=GuardrailEventHooks.pre_call
)
is False
)
# Test 3: guardrail in the opt-out list inside metadata → skipped
data_metadata = { data_metadata = {
"model": "gpt-3.5-turbo", "model": "gpt-3.5-turbo",
"messages": [{"role": "user", "content": "test"}], "messages": [{"role": "user", "content": "test"}],
@ -291,7 +260,7 @@ class TestCustomGuardrailShouldRunGuardrail:
custom_guardrail.should_run_guardrail( custom_guardrail.should_run_guardrail(
data=data_metadata, event_type=GuardrailEventHooks.pre_call data=data_metadata, event_type=GuardrailEventHooks.pre_call
) )
is False is True
) )
# Test 4: a different guardrail in the opt-out list → still runs # Test 4: a different guardrail in the opt-out list → still runs
@ -588,7 +557,9 @@ class TestGuardrailSensitiveFieldStripping:
duration=1.0, duration=1.0,
) )
logged_response = request_data["metadata"]["standard_logging_guardrail_information"][0]["guardrail_response"] logged_response = request_data["metadata"][
"standard_logging_guardrail_information"
][0]["guardrail_response"]
assert "secret_fields" not in logged_response assert "secret_fields" not in logged_response
assert "sk-live-SHOULD-NOT-APPEAR" not in json.dumps(logged_response) assert "sk-live-SHOULD-NOT-APPEAR" not in json.dumps(logged_response)
@ -599,7 +570,12 @@ class TestGuardrailSensitiveFieldStripping:
guardrail.add_standard_logging_guardrail_information_to_request_data( guardrail.add_standard_logging_guardrail_information_to_request_data(
guardrail_json_response=[ guardrail_json_response=[
{"result": "ok", "secret_fields": {"raw_headers": {"authorization": "Bearer sk-secret"}}}, {
"result": "ok",
"secret_fields": {
"raw_headers": {"authorization": "Bearer sk-secret"}
},
},
{"result": "also_ok"}, {"result": "also_ok"},
], ],
request_data=request_data, request_data=request_data,
@ -608,6 +584,7 @@ class TestGuardrailSensitiveFieldStripping:
) )
import json import json
serialized = json.dumps(request_data) serialized = json.dumps(request_data)
assert "secret_fields" not in serialized assert "secret_fields" not in serialized
assert "sk-secret" not in serialized assert "sk-secret" not in serialized
@ -621,21 +598,21 @@ class TestCustomGuardrailPassthroughSupport:
""" """
Test that async_post_call_success_deployment_hook handles raw httpx.Response objects Test that async_post_call_success_deployment_hook handles raw httpx.Response objects
from passthrough endpoints without crashing with TypeError. from passthrough endpoints without crashing with TypeError.
This tests Fix #3: TypeError: TypedDict does not support instance and class checks This tests Fix #3: TypeError: TypedDict does not support instance and class checks
""" """
import httpx import httpx
custom_guardrail = CustomGuardrail() custom_guardrail = CustomGuardrail()
# Mock the async_post_call_success_hook to return None (guardrail didn't modify response) # Mock the async_post_call_success_hook to return None (guardrail didn't modify response)
custom_guardrail.async_post_call_success_hook = AsyncMock(return_value=None) custom_guardrail.async_post_call_success_hook = AsyncMock(return_value=None)
# Create a mock httpx.Response object (typical passthrough response) # Create a mock httpx.Response object (typical passthrough response)
mock_response = AsyncMock(spec=httpx.Response) mock_response = AsyncMock(spec=httpx.Response)
mock_response.status_code = 200 mock_response.status_code = 200
mock_response.text = "Mock response" mock_response.text = "Mock response"
request_data = { request_data = {
"guardrails": ["test_guardrail"], "guardrails": ["test_guardrail"],
"user_api_key_user_id": "test_user", "user_api_key_user_id": "test_user",
@ -644,14 +621,14 @@ class TestCustomGuardrailPassthroughSupport:
"user_api_key_hash": "test_hash", "user_api_key_hash": "test_hash",
"user_api_key_request_route": "passthrough_route", "user_api_key_request_route": "passthrough_route",
} }
# This should not raise TypeError: TypedDict does not support instance and class checks # This should not raise TypeError: TypedDict does not support instance and class checks
result = await custom_guardrail.async_post_call_success_deployment_hook( result = await custom_guardrail.async_post_call_success_deployment_hook(
request_data=request_data, request_data=request_data,
response=mock_response, response=mock_response,
call_type=CallTypes.allm_passthrough_route, call_type=CallTypes.allm_passthrough_route,
) )
# When result is None, should return the original response # When result is None, should return the original response
assert result == mock_response assert result == mock_response
@ -659,53 +636,53 @@ class TestCustomGuardrailPassthroughSupport:
async def test_async_post_call_success_deployment_hook_with_none_call_type(self): async def test_async_post_call_success_deployment_hook_with_none_call_type(self):
""" """
Test that async_post_call_success_deployment_hook handles None call_type gracefully. Test that async_post_call_success_deployment_hook handles None call_type gracefully.
This ensures that even if call_type is None (before fix #1), the guardrail doesn't crash. This ensures that even if call_type is None (before fix #1), the guardrail doesn't crash.
""" """
custom_guardrail = CustomGuardrail() custom_guardrail = CustomGuardrail()
# Mock the async_post_call_success_hook to return None # Mock the async_post_call_success_hook to return None
custom_guardrail.async_post_call_success_hook = AsyncMock(return_value=None) custom_guardrail.async_post_call_success_hook = AsyncMock(return_value=None)
mock_response = AsyncMock() mock_response = AsyncMock()
request_data = { request_data = {
"guardrails": ["test_guardrail"], "guardrails": ["test_guardrail"],
"user_api_key_user_id": "test_user", "user_api_key_user_id": "test_user",
} }
# Call with None call_type - should not crash # Call with None call_type - should not crash
result = await custom_guardrail.async_post_call_success_deployment_hook( result = await custom_guardrail.async_post_call_success_deployment_hook(
request_data=request_data, request_data=request_data,
response=mock_response, response=mock_response,
call_type=None, call_type=None,
) )
# Should return the original response when result is None # Should return the original response when result is None
assert result == mock_response assert result == mock_response
def test_is_valid_response_type_with_none(self): def test_is_valid_response_type_with_none(self):
""" """
Test _is_valid_response_type helper method correctly identifies None as invalid. Test _is_valid_response_type helper method correctly identifies None as invalid.
This is part of Fix #3: Safely handling TypedDict types that don't support isinstance checks. This is part of Fix #3: Safely handling TypedDict types that don't support isinstance checks.
""" """
custom_guardrail = CustomGuardrail() custom_guardrail = CustomGuardrail()
# None should be invalid # None should be invalid
assert custom_guardrail._is_valid_response_type(None) is False assert custom_guardrail._is_valid_response_type(None) is False
def test_is_valid_response_type_with_typeddict_error(self): def test_is_valid_response_type_with_typeddict_error(self):
""" """
Test _is_valid_response_type gracefully handles TypeError from TypedDict. Test _is_valid_response_type gracefully handles TypeError from TypedDict.
This tests Fix #3: When isinstance() is called with TypedDict types, it raises TypeError. This tests Fix #3: When isinstance() is called with TypedDict types, it raises TypeError.
The method should catch this and allow the response through. The method should catch this and allow the response through.
""" """
from litellm.types.utils import ModelResponse from litellm.types.utils import ModelResponse
custom_guardrail = CustomGuardrail() custom_guardrail = CustomGuardrail()
# Create a valid LiteLLM response object # Create a valid LiteLLM response object
response = ModelResponse( response = ModelResponse(
id="test-id", id="test-id",
@ -714,13 +691,12 @@ class TestCustomGuardrailPassthroughSupport:
model="test-model", model="test-model",
object="chat.completion", object="chat.completion",
) )
# This should return True (it's a valid response type or TypeError is caught) # This should return True (it's a valid response type or TypeError is caught)
result = custom_guardrail._is_valid_response_type(response) result = custom_guardrail._is_valid_response_type(response)
assert result is True assert result is True
class TestEventTypeLogging: class TestEventTypeLogging:
"""Tests for event_type logging in guardrail information.""" """Tests for event_type logging in guardrail information."""
@ -1014,7 +990,9 @@ class TestTracingFieldsPopulation:
guardrail_json_response="blocked", guardrail_json_response="blocked",
request_data=request_data, request_data=request_data,
guardrail_status="guardrail_intervened", guardrail_status="guardrail_intervened",
tracing_detail=GuardrailTracingDetail(policy_template="EU AI Act Article 5"), tracing_detail=GuardrailTracingDetail(
policy_template="EU AI Act Article 5"
),
) )
slg_list = request_data["metadata"]["standard_logging_guardrail_information"] slg_list = request_data["metadata"]["standard_logging_guardrail_information"]